Skip to content

Commit

Permalink
Initial implementation of Linked Alerts
Browse files Browse the repository at this point in the history 
When the cot_uuid matches and existing CoT UUID a new CoT event is built and a link tag is added to the new alert Cot
  • Loading branch information
Keith Williams committed Apr 15, 2021
1 parent e251ca1 commit d60d084
Show file tree
Hide file tree
Showing 6 changed files with 142 additions and 36 deletions.
1 change: 1 addition & 0 deletions TAK Server Plugin/Plugin/src/takserver-sender-plugin-mcs/build.gradle
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ dependencies {
compile group: 'com.google.code.gson', name: 'gson', version: '2.8.6'
implementation group: 'com.rabbitmq', name: 'amqp-client', version: '5.6.0'
compile group: 'org.json', name: 'json', version: '20201115'
compile group: 'org.apache.httpcomponents.client5', name: 'httpclient5', version: '5.0'
}


30 changes: 27 additions & 3 deletions ...gin/src/takserver-sender-plugin-mcs/src/main/java/tak/server/plugins/McsSenderPlugin.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
import tak.server.plugins.dto.EntityDto;
import tak.server.plugins.interfaces.MessageCallback;
import tak.server.plugins.messagebroker.RabbitMQClient;
import tak.server.plugins.missionapi.TakServerCoTApi;
import tak.server.plugins.missionapi.TakServerCoTCallback;
import tak.server.plugins.processing.MessageConsumer;
import tak.server.plugins.processing.MessageProducer;
import tak.server.plugins.processing.ProcessingMessage;
Expand All @@ -48,7 +50,7 @@ public class McsSenderPlugin extends MessageSenderBase implements MessageCallbac
private MessageConsumer _messageConsumer;
private BlockingQueue<ProcessingMessage> _blockingQueue;
private ExecutorService _executor = Executors.newFixedThreadPool(2);

public static Boolean VerboseLogging = false;

@SuppressWarnings("unchecked")
Expand Down Expand Up @@ -116,7 +118,29 @@ public void messageReceived(String topic, String message){
return;
}

takMessage = McsCoTConverter.convertToMessage(event, config);
//TODO - park it in prefs
String targetAddress = "127.0.0.1";
int targetPort = 8080;
TakServerCoTApi.queryForCotEvent(targetAddress, targetPort, event.getUid(),
new TakServerCoTCallback(){
@Override
public void cotResult(Boolean success, String cot) {
try {
Message eventMessage = null;
if (success) {
final Message parentMessage = getConverter().cotStringToDataMessage(cot, null, Integer.toString(System.identityHashCode(this)));
eventMessage = McsCoTConverter.convertToLinkedMessage(event, parentMessage.getPayload().getCotEvent(), config);
}
else
eventMessage = McsCoTConverter.convertToMessage(event, config);

send(eventMessage);
}
catch (Exception exception) {
_logger.error("error converting to linked message", exception);
}
}
});
}
else {
EntityDto entity = McsCoTConverter.convertToEntity(message, config);
Expand All @@ -126,6 +150,7 @@ public void messageReceived(String topic, String message){
}

takMessage = McsCoTConverter.convertToMessage(entity, config);
send(takMessage);
}

if (takMessage == null){
Expand All @@ -136,7 +161,6 @@ public void messageReceived(String topic, String message){
if(VerboseLogging)
_logger.info("TAK message converted: " + takMessage);

send(takMessage);
} catch (Exception exception) {
_logger.error("error converting message ", message, exception);
}
Expand Down
61 changes: 61 additions & 0 deletions ...server-sender-plugin-mcs/src/main/java/tak/server/plugins/missionapi/TakServerCoTApi.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
package tak.server.plugins.missionapi;

import java.util.concurrent.Future;

import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
import org.apache.hc.client5.http.async.methods.SimpleHttpRequests;
import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.reactor.IOReactorConfig;
import org.apache.hc.core5.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TakServerCoTApi {
private static final Logger logger = LoggerFactory.getLogger(TakServerCoTApi.class);

public static void queryForCotEvent(String targetAddress, int targetPort, String uuid, TakServerCoTCallback callback) {
try {
final IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
.setSoTimeout(Timeout.ofSeconds(5))
.build();

final CloseableHttpAsyncClient client = HttpAsyncClients.custom()
.setIOReactorConfig(ioReactorConfig)
.build();

client.start();

String requestUri = "/Marti/api/cot/xml/" + uuid;
final HttpHost target = new HttpHost(targetAddress, targetPort);
final SimpleHttpRequest httpget = SimpleHttpRequests.get(target, requestUri);
logger.info("Executing request " + httpget.getMethod() + " " + httpget.getUri());
final Future<SimpleHttpResponse> future = client.execute(
httpget,
new FutureCallback<SimpleHttpResponse>() {
@Override
public void completed(final SimpleHttpResponse response) {
callback.cotResult(response.getCode() == 200, response.getBody().getBodyText());
}

@Override
public void failed(final Exception ex) {
logger.error(target.toURI() + requestUri + "->" + ex);
callback.cotResult(false, "");
}

@Override
public void cancelled() {
logger.error(target.toURI() + requestUri + " cancelled");
callback.cotResult(false, "");
}
});
}
catch (Exception e) {
logger.info("exception making HTTP request", e);
}
}
}
5 changes: 5 additions & 0 deletions ...r-sender-plugin-mcs/src/main/java/tak/server/plugins/missionapi/TakServerCoTCallback.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
package tak.server.plugins.missionapi;

public interface TakServerCoTCallback {
void cotResult(Boolean success, String cot);
}
64 changes: 37 additions & 27 deletions ...kserver-sender-plugin-mcs/src/main/java/tak/server/plugins/utilities/McsCoTConverter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,39 @@ public class McsCoTConverter {

public static Message convertToMessage(EventDto event, PluginConfiguration configuration) {
Message.Builder messageBuilder = MessageOuterClass.Message.newBuilder();
TakMessage.Builder payloadBuilder = messageBuilder.getPayloadBuilder();
buildEventMessage(event, messageBuilder, configuration);

return messageBuilder.build();
}

public static Message convertToLinkedMessage(EventDto event, CotEvent linkedCotEvent, PluginConfiguration configuration) {
Message.Builder messageBuilder = MessageOuterClass.Message.newBuilder();
buildEventMessage(event, messageBuilder, configuration);

TakMessage.Builder payloadBuilder = messageBuilder.getPayloadBuilder();
CotEvent.Builder cotEventBuilder = payloadBuilder.getCotEventBuilder();
DetailOuterClass.Detail.Builder detailBuilder = cotEventBuilder.getDetailBuilder();

Instant instant = Instant.now();
Long timeMs = instant.toEpochMilli() ;
cotEventBuilder.setUid(event.getUid() + "-alert-" + timeMs.toString());

cotEventBuilder.setLat(linkedCotEvent.getLat());
cotEventBuilder.setLon(linkedCotEvent.getLon());

String existingDetail = detailBuilder.getXmlDetail();
String linkDetail = "";
//Keep it simple for now
String linkXML = "<link relation=\"p-p\" type=\"PARENT_TYPE\" uid=\"PARENT_UUID\"/>"
.replace("PARENT_TYPE", linkedCotEvent.getType())
.replace("PARENT_UUID", linkedCotEvent.getUid());
detailBuilder.setXmlDetail(existingDetail + linkXML);

return messageBuilder.build();
}

public static void buildEventMessage(EventDto event, Message.Builder messageBuilder, PluginConfiguration configuration) {
TakMessage.Builder payloadBuilder = messageBuilder.getPayloadBuilder();
CotEvent.Builder cotEventBuilder = payloadBuilder.getCotEventBuilder();
DetailOuterClass.Detail.Builder detailBuilder = cotEventBuilder.getDetailBuilder();

Expand All @@ -49,34 +81,14 @@ public static Message convertToMessage(EventDto event, PluginConfiguration confi
Long timeMs = instant.toEpochMilli() ;
Long staleMs = timeMs + (5 * 60 * 1000); //Five minutes

cotEventBuilder.setType("a-u");
cotEventBuilder.setHow("m-r");
cotEventBuilder.setType("b-m-p-s-p-i");//Bits-mapping-designated point-sensor-point-interest
cotEventBuilder.setHow("m-r");//Machine-relayed

cotEventBuilder.setSendTime(timeMs);
cotEventBuilder.setStartTime(timeMs);
cotEventBuilder.setStaleTime(staleMs);

if(configuration.containsProperty("simulateAlertLocation")
&& (boolean)configuration.getProperty("simulateAlertLocation") == true) {
if (event.getPoint().getLat() == 0.0 || event.getPoint().getLon() == 0.0) {
//27.6615493 -81.2769707 - Generally around Avon Park
Random r = new Random();
double tempLat = 27.6615493 + r.nextDouble() * .08;
double tempLon = -81.2769707 + r.nextDouble() * .08;

cotEventBuilder.setLat(tempLat);
cotEventBuilder.setLon(tempLon);
}
else {
cotEventBuilder.setLat(event.getPoint().getLat());
cotEventBuilder.setLon(event.getPoint().getLon());
}
}
else {
cotEventBuilder.setLat(event.getPoint().getLat());
cotEventBuilder.setLon(event.getPoint().getLon());
}

cotEventBuilder.setLat(event.getPoint().getLat());
cotEventBuilder.setLon(event.getPoint().getLon());

cotEventBuilder.setHae(9999999);
cotEventBuilder.setCe(9999999);
Expand All @@ -90,8 +102,6 @@ public static Message convertToMessage(EventDto event, PluginConfiguration confi
detailJsonObject.put(FROM_MCS, "true");
String xmlDetailData = XML.toString(detailJsonObject);
detailBuilder.setXmlDetail(xmlDetailData);

return messageBuilder.build();
}

public static Message convertToMessage(EntityDto entity, PluginConfiguration configuration) {
Expand Down
17 changes: 11 additions & 6 deletions TAK Server Plugin/RabbitMQ Client/Program.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,13 @@ class Program

private static string MESSAGE = "{\"uid\":\"CustomIDForTAKServerTest.Entity\",\"type\":\"a-f-G-U-C\",\"time\":\"1614187736429\",\"start\":\"1614187736429\",\"stale\":\"1614191352000\",\"how\":\"m-g\",\"point\":{\"lat\":\"39.0495\",\"lon\":\"-85.7445\",\"hae\":\"9999999\",\"ce\":\"9999999\",\"le\":\"9999999\"},\"detail\":{\"milsym2525C\":\"SFGPUCI*****\", \"video\":\"https://cdn.bitdegree.org/learn/Pexels%20Videos%203373.mp4?raw=true\", \"image\": \"IMAGEPLACEHOLDER\", \"feedType\":\"OSH_SENSOR\",\"serviceDetails\":{\"serviceUrl\":\"THE SERVICE URL\",\"provider\":{\"providerName\":\"SOME NAME\",\"providerData\":\"SOME DATA\"},\"offeringID\":\"THE OFFERING ID\",\"temporalData\":\"THE TEMPORAL DATA\"}}}";

private static string EVENT_MESSAGE = "{\"cot_uid\":\"CustomIDForTAKServerTest.Alert\",\"message\":\"This is a demo alert from a test client (no location)\",\"type\":\"info\"}";
//private static string EVENT_MESSAGE = "{\"cot_uid\":\"CustomIDForTAKServerTest.Alert\",\"message\":\"This is a demo alert from a test client (no location)\",\"type\":\"info\"}";

private static string EVENT_MESSAGE = "{\"cot_uid\":\"CustomIDForTAKServerTest.Entity\",\"message\":\"This is a demo alert from a test client at TIME\",\"type\":\"info\"}";
//ANDROID-1e159fc8250eb070
//private static string EVENT_MESSAGE = "{\"cot_uid\":\"ANDROID-1e159fc8250eb070\",\"message\":\"This is a demo alert from a test client (no location)\",\"type\":\"info\"}";

private static string EVENT_MESSAGE_WITH_LOCATION = "{\"cot_uid\":\"CustomIDForTAKServerTest.Alert\",\"message\":\"This is a demo alert from a test client\",\"type\":\"info\",\"point\":{\"lat\":\"30.4299511\",\"lon\":\"-86.6061093\",\"hae\":\"9999999\",\"ce\":\"9999999\",\"le\":\"9999999\"}}";
private static string EVENT_MESSAGE_WITH_LOCATION = "{\"cot_uid\":\"CustomIDForTAKServerTest.Entity\",\"message\":\"This is a demo alert from a test client\",\"type\":\"info\",\"point\":{\"lat\":\"30.4299511\",\"lon\":\"-86.6061093\",\"hae\":\"9999999\",\"ce\":\"9999999\",\"le\":\"9999999\"}}";

private static string IMAGE_URL = @"https://i.picsum.photos/id/1025/4951/3301.jpg?hmac=_aGh5AtoOChip_iaMo8ZvvytfEojcgqbCH7dzaz-H8Y";

Expand Down Expand Up @@ -66,7 +70,7 @@ static void Main(string[] args)
body: body);

Console.WriteLine(" [x] Sent {0}", message);
await Task.Delay(10000);
await Task.Delay(5000);
}
}
catch (Exception ex)
Expand All @@ -82,15 +86,16 @@ static void Main(string[] args)
{
while (true)
{
var body = Encoding.UTF8.GetBytes(EVENT_MESSAGE);
var message = EVENT_MESSAGE.Replace("TIME", DateTime.UtcNow.ToString("HH:mm:ss"));
var body = Encoding.UTF8.GetBytes(message);

channel.BasicPublish(exchange: EXCHANGE,
routingKey: EVENT_ROUTING_KEY,
basicProperties: null,
body: body);

Console.WriteLine(" [x] Sent {0}", EVENT_MESSAGE);
await Task.Delay(5000);
Console.WriteLine(" [x] Sent {0}", message);
await Task.Delay(20000);
}
}
catch (Exception ex)
Expand Down

0 comments on commit d60d084

Please sign in to comment.