Telematic bridge nats

.devcontainer/docker-compose-vscode.yml

@@ -24,7 +24,8 @@ services:
       - SIM_V2X_PORT=5757
       - SIM_INTERACTION_PORT=7576
       - V2X_PORT=8686
-      - INFRASTRUCTURE_ID=1
+      - INFRASTRUCTURE_ID=rsu_<J2735 MAP MESSAGE INTERSECTION ID>
+      - INFRASTRUCTURE_NAME=<RSU_NAME>
       - SENSOR_JSON_FILE_PATH=/var/www/plugins/MAP/sensors.json
     secrets:
       - mysql_password

configuration/amd64/docker-compose.yml

@@ -38,6 +38,8 @@ services:
       - db
     environment:
       - MYSQL_PASSWORD=/run/secrets/mysql_password
+      - INFRASTRUCTURE_ID=rsu_<J2735 MAP MESSAGE INTERSECTION ID>
+      - INFRASTRUCTURE_NAME=<RSU_NAME>
     secrets:
       - mysql_password
     volumes:

src/tmx/TmxUtils/src/simulation/SimulationEnvUtils.h

@@ -66,6 +66,12 @@ namespace tmx::utils::sim{
          * for CDASim connection.
          */
         constexpr inline static const char *INFRASTRUCTURE_ID = "INFRASTRUCTURE_ID";
+
+        /**
+         * @brief Name of environment variable for storing infrastructure name of v2xhub. Only necessary in SIMULATION MODE 
+         * for CDASim connection.
+        */
+        constexpr inline static const char *INFRASTRUCTURE_NAME = "INFRASTRUCTURE_NAME";
         /**
          * @brief Function to return bool indicating whether V2X-Hub deployment is in SIMULATION MODE or not.
          * @return true if SIMULATION_MODE is "true" or "TRUE" and false otherwise.

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,16 @@
             "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": "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,16 @@
 #include "TelematicBridgePlugin.h"
 
+using namespace tmx::utils;
+using namespace std;
+
 namespace TelematicBridge
 {
     TelematicBridgePlugin::TelematicBridgePlugin(const string &name) : PluginClient(name)
     {
+        _telematicUnitPtr = make_unique<TelematicUnit>();
+        _unitId = std::getenv("INFRASTRUCTURE_ID");
+        _unitName = std::getenv("INFRASTRUCTURE_NAME");
+        UpdateConfigSettings();
         AddMessageFilter("*", "*", IvpMsgFlags_None);
         AddMessageFilter("J2735", "*", IvpMsgFlags_RouteDSRC);
         SubscribeToMessages();
@@ -20,14 +27,53 @@ 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>("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,30 @@
 
 #include "PluginClient.h"
 #include "TelematicBridgeMsgWorker.h"
+#include "TelematicUnit.h"
+#include <simulation/SimulationEnvUtils.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::unique_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;
     };