Telematic bridge nats

src/v2i-hub/TelematicBridgePlugin/CMakeLists.txt

@@ -3,13 +3,15 @@ PROJECT (TelematicBridgePlugin VERSION 7.5.1 LANGUAGES CXX)
 set (TMX_PLUGIN_NAME "Telematic Bridge")
 
 BuildTmxPlugin()
-TARGET_LINK_LIBRARIES ( ${PROJECT_NAME} tmxutils jsoncpp)
+TARGET_LINK_LIBRARIES ( ${PROJECT_NAME} tmxutils jsoncpp nats)
 
 ####################################################
 ##################  Testing  #######################
 ####################################################
+add_library(${PROJECT_NAME}_lib src/TelematicUnit.cpp)
+target_link_libraries(${PROJECT_NAME}_lib PUBLIC  tmxutils jsoncpp nats)
 enable_testing()
 include_directories(${PROJECT_SOURCE_DIR}/src)
 file(GLOB_RECURSE TEST_SOURCES LIST_DIRECTORIES false test/*.h test/*.cpp)
 add_executable(${PROJECT_NAME}_test ${TEST_SOURCES})
-target_link_libraries(${PROJECT_NAME}_test PRIVATE gtest tmxutils jsoncpp)
+target_link_libraries(${PROJECT_NAME}_test PRIVATE gtest ${PROJECT_NAME}_lib)

src/v2i-hub/TelematicBridgePlugin/README.md

@@ -0,0 +1,96 @@
+## Telematic bridge introduction
+Telematic bridge is a V2xHub plugin used to collect data stream generated by V2xHub and send the data stream into [telematic tool](https://github.com/usdot-fhwa-stol/cda-telematics) hosted in the AWS cloud. 
+## NATS Publisher/Subscriber
+### NATS Connections and registration
+#### Telematic plugin sends registration request to telematic server
+##### Subject
+NATS subject: *.register_unit
+##### Request
+```
+{
+  "unit_id": "<unit_id>"
+}
+```
+##### Response
+```
+{
+  "unit_id": "<unit_id>",
+  "unit_type": "infrastructure",
+  "unit_name": "East Intersection",
+  "timestamp": "1678998191815233965",
+  "event_name": "wfd_integration_testing",
+  "location": "TFHRC",
+  "testing_type": "Integration"
+}
+```
+
+### Available topics
+#### Telematic plugin receives request from telematic UI
+##### Subject
+NATS subject: <unit_id>.available_topics
+##### Request 
+```
+{
+  "unit_id": "<unit_id>"
+}
+```
+##### Reply
+
+```
+ {
+  "unit_id": "<unit_id>",
+  "unit_type": "infrastructure",
+  "unit_name": "East Intersection",
+  "timestamp": "1678998191815233965",
+  "event_name": "wfd_integration_testing",
+  "location": "TFHRC",
+  "testing_type": "Integration",
+  "topics": [
+    {
+      "name": "J2735_TMSG03-P_CARMAStreetsPlugin"
+    },
+    {
+      "name": "<topic_name_2>"
+    }
+  ]
+}
+```
+
+### Selected topics
+#### Telematic plugin receives selected topics from telematic UI
+##### Subject
+NATS subject: <unit_id>.publish_topics
+#### Request
+```
+{
+  "unit_id": "<unit_id>",
+  "unit_type": "infrastructure",
+  "timestamp": 1663084528513000400,
+  "event_name": "wfd_integration_testing",
+  "location": "TFHRC",
+  "testing_type": "Integration",
+  "topics": [
+    "<topic_name_1>",
+    "<topic_name_2>"
+  ]
+}
+```
+##### Reply
+```
+"request received!"
+```
+
+## Check status
+#### Telematic plugin receives live status request from telematic server
+##### Subject
+NATS subject: <unit_id>.check_status
+##### Request
+```
+{
+  "unit_id": "<unit_id>"
+}
+```
+##### Reponse
+```
+"OK"
+```

src/v2i-hub/TelematicBridgePlugin/manifest.json

@@ -11,6 +11,26 @@
             "key": "LogLevel",
             "default": "INFO",
             "description": "The log level for this plugin"
+        },
+        {
+            "key": "NATSUrl",
+            "default": "nats://127.0.0.1:4222",
+            "description": "The NATS connection URL"            
+        },
+        {
+            "key": "UnitId",
+            "default": "v2xhub_id",
+            "description": "The unique identifier for a telematic unit"            
+        },
+        {
+            "key": "UnitName",
+            "default": "East Intersection",
+            "description": "A descriptive name for each telematic unit. It can also be the intersection name where this unit collect data from."            
+        },
+        {
+            "key": "MessageExclusionList",
+            "default": "System_KeepAlive_CommandPlugin,System_KeepAlive_CARMAStreetsPlugin,System_KeepAlive_CDASimAdapter,System_KeepAlive_MessageReceiver",
+            "description": "The list of messages are excluded from the available message list. Message name is a combination of message type, subtype and source separated by underscore. E.G: type_subtype_source"            
         }
     ]
 }

src/v2i-hub/TelematicBridgePlugin/src/TelematicBridgePlugin.cpp

@@ -1,9 +1,14 @@
 #include "TelematicBridgePlugin.h"
 
+using namespace tmx::utils;
+using namespace std;
+
 namespace TelematicBridge
 {
     TelematicBridgePlugin::TelematicBridgePlugin(const string &name) : PluginClient(name)
     {
+        _telematicUnitPtr = make_shared<TelematicUnit>();
+        UpdateConfigSettings();
         AddMessageFilter("*", "*", IvpMsgFlags_None);
         AddMessageFilter("J2735", "*", IvpMsgFlags_RouteDSRC);
         SubscribeToMessages();
@@ -20,14 +25,55 @@ namespace TelematicBridge
                 auto messageFm = (MessageFrame_t *)calloc(1, sizeof(MessageFrame_t));
                 DecodeJ2735Msg(msg->payload->valuestring, messageFm);
                 string xml_payload_str = ConvertJ2735FrameToXML(messageFm);
-                ASN_STRUCT_FREE(asn_DEF_MessageFrame, messageFm);                
-                json["payload"] = StringToJson(xml2Json(xml_payload_str));
+                ASN_STRUCT_FREE(asn_DEF_MessageFrame, messageFm);
+                string json_payload_str = xml2Json(xml_payload_str.c_str());
+                json["payload"] = StringToJson(json_payload_str);
             }
 
-            auto jsonStr = JsonToString(json);
-            PLOG(logINFO) << jsonStr;
+            stringstream topic;
+            topic << (msg->type ? msg->type : "") << "_" << (msg->subtype ? msg->subtype : "") << "_" << (msg->source ? msg->source : "");
+            auto topicStr = topic.str();
+            _telematicUnitPtr->updateAvailableTopics(topicStr);
+            if (_telematicUnitPtr->inSelectedTopics(topicStr))
+            {
+                _telematicUnitPtr->publishMessage(topicStr, json);
+            }
+        }
+    }
+
+    void TelematicBridgePlugin::UpdateConfigSettings()
+    {
+        lock_guard<mutex> lock(_configMutex);
+        GetConfigValue<string>("NATSUrl", _natsURL);
+        GetConfigValue<string>("UnitId", _unitId);
+        GetConfigValue<string>("UnitName", _unitName);
+        GetConfigValue<string>("MessageExclusionList", _excludedMessages);
+        unit_st unit = {_unitId, _unitName, UNIT_TYPE_INFRASTRUCTURE};
+        if (_telematicUnitPtr)
+        {
+            _telematicUnitPtr->setUnit(unit);
+            _telematicUnitPtr->updateExcludedTopics(_excludedMessages);
+        }
+    }
+
+    void TelematicBridgePlugin::OnStateChange(IvpPluginState state)
+    {
+        PluginClient::OnStateChange(state);
+        if (state == IvpPluginState_registered)
+        {
+            UpdateConfigSettings();
+            if (_telematicUnitPtr)
+            {
+                _telematicUnitPtr->connect(_natsURL);
+            }
         }
     }
+
+    void TelematicBridgePlugin::OnConfigChanged(const char *key, const char *value)
+    {
+        PluginClient::OnConfigChanged(key, value);
+        UpdateConfigSettings();
+    }
 }
 
 // The main entry point for this application.

src/v2i-hub/TelematicBridgePlugin/src/TelematicBridgePlugin.h

@@ -3,21 +3,29 @@
 
 #include "PluginClient.h"
 #include "TelematicBridgeMsgWorker.h"
+#include "TelematicUnit.h"
+
 
-using namespace tmx::utils;
-using namespace std;
 namespace TelematicBridge
 {
-
-
     class TelematicBridgePlugin : public tmx::utils::PluginClient
     {
     private:
         static CONSTEXPR const char *Telematic_MSGTYPE_J2735_STRING = "J2735";
+        static CONSTEXPR const char *UNIT_TYPE_INFRASTRUCTURE = "Infrastructure";
+        std::shared_ptr<TelematicUnit> _telematicUnitPtr;
+        std::string _unitId;
+        std::string _unitName;
+        std::string _natsURL;
+        std::string _excludedMessages;
+        std::mutex _configMutex;
         void OnMessageReceived(IvpMessage *msg);
 
     public:
-        explicit TelematicBridgePlugin(const string& name);
+        explicit TelematicBridgePlugin(const std::string &name);
+        void OnConfigChanged(const char *key, const char *value) override;
+        void OnStateChange(IvpPluginState state) override;
+        void UpdateConfigSettings();
         ~TelematicBridgePlugin() override = default;
     };