v2x hub Integration: Implement telematic plugin to subscribe to all TMX messages from V2xHub

.sonarqube/sonar-scanner.properties

@@ -56,7 +56,8 @@ sonar.modules=  PedestrianPlugin, \
   CARMAStreetsPlugin, \
   ERVCloudForwardingPlugin, \
   CDASimAdapter, \
-  RSUHealthMonitorPlugin
+  RSUHealthMonitorPlugin, \
+  TelematicBridgePlugin
 
 
 
@@ -84,6 +85,7 @@ CARMAStreetsPlugin.sonar.projectBaseDir         =src/v2i-hub/CARMAStreetsPlugin
 ERVCloudForwardingPlugin.sonar.projectBaseDir         =src/v2i-hub/ERVCloudForwardingPlugin
 CDASimAdapter.sonar.projectBaseDir         =src/v2i-hub/CDASimAdapter
 RSUHealthMonitorPlugin.sonar.projectBaseDir     =src/v2i-hub/RSUHealthMonitorPlugin
+TelematicBridgePlugin.sonar.projectBaseDir     =src/v2i-hub/TelematicBridgePlugin
 
 
 
@@ -121,6 +123,8 @@ CDASimAdapter.sonar.sources          =src
 CDASimAdapter.sonar.exclusions                  =test/**
 RSUHealthMonitorPlugin.sonar.sources    =src
 RSUHealthMonitorPlugin.sonar.exclusions =test/**
+TelematicBridgePlugin.sonar.sources    =src
+TelematicBridgePlugin.sonar.exclusions =test/**
 
 # Tests
 # Note: For C++ setting this field does not cause test analysis to occur. It only allows the test source code to be evaluated.
@@ -149,3 +153,4 @@ CARMAStreetsPlugin.sonar.tests=test
 ERVCloudForwardingPlugin.sonar.tests=test
 CDASimAdapter.sonar.tests=test
 RSUHealthMonitorPlugin.sonar.tests=test
+TelematicBridgePlugin.sonar.tests=test

ext/build.sh

@@ -55,4 +55,14 @@ cd NemaTode
 cmake .
 make -j${numCPU}
 make install
-popd
+popd
+
+# Nats C API
+pushd /tmp
+git clone https://github.com/nats-io/nats.c
+cd nats.c
+cmake . -DNATS_BUILD_NO_SPIN=ON
+make -j${numCPU}
+make install
+popd
+

scripts/install_dependencies.sh

@@ -31,6 +31,9 @@ DEPENDENCIES="build-essential \
     wget \
     zip \
     zlib1g \
+    rapidjson-dev \
+    librapidxml-dev \
+    libprotobuf-c-dev \
     curl"
 
 # STOL library dependencies

src/v2i-hub/RSUHealthMonitorPlugin/src/RSUHealthMonitorWorker.cpp

@@ -145,7 +145,11 @@ namespace RSUHealthMonitor
                     }
                     else if (success)
                     {
-                        rsuStatuJson.append(populateJson(config.field, responseVal));
+                        auto json = populateJson(config.field, responseVal);
+                        for(const auto &key: json.getMemberNames())
+                        {
+                            rsuStatuJson[key] = json[key];
+                        }
                     }
                 }
             }

src/v2i-hub/RSUHealthMonitorPlugin/test/test_RSUHealthMonitorWorker.cpp

@@ -87,19 +87,22 @@ namespace RSUHealthMonitor
         double expected_longitude = -77.1496;
         ASSERT_NEAR(expected_latitude, json["rsuGpsOutputStringLatitude"].asDouble(), 0.001);
         ASSERT_NEAR(expected_longitude, json["rsuGpsOutputStringLongitude"].asDouble(), 0.001);
-        rsuStatusJson.append(json);
+        for(const auto& key: json.getMemberNames())
+        {
+            rsuStatusJson[key] = json[key];
+        }
 
         snmp_response_obj intObj;
         intObj.type = snmp_response_obj::response_type::INTEGER;
         intObj.val_int = 4;
 
         json = _rsuWorker->populateJson("rsuMode", intObj);
         ASSERT_EQ(4, json["rsuMode"].asInt64());
-        rsuStatusJson.append(json);
+        rsuStatusJson["rsuMode"] = json["rsuMode"];
 
         Json::FastWriter fasterWirter;
         string json_str = fasterWirter.write(rsuStatusJson);
-        string expectedStr = "[{\"rsuGpsOutputString\":\"$GPGGA,142440.00,3857.3065,N,07708.9734,W,2,18,0.65,86.18,M,-34.722,M,,*62\",\"rsuGpsOutputStringLatitude\":38.955108330000002,\"rsuGpsOutputStringLongitude\":-77.149556669999996},{\"rsuMode\":4}]\n";
+        string expectedStr = "{\"rsuGpsOutputString\":\"$GPGGA,142440.00,3857.3065,N,07708.9734,W,2,18,0.65,86.18,M,-34.722,M,,*62\",\"rsuGpsOutputStringLatitude\":38.955108330000002,\"rsuGpsOutputStringLongitude\":-77.149556669999996,\"rsuMode\":4}\n";
         ASSERT_EQ(expectedStr, json_str);
         ASSERT_EQ(4, _rsuWorker->getJsonKeys(rsuStatusJson).size());
         ASSERT_EQ(1, _rsuWorker->getJsonKeys(json).size());

src/v2i-hub/SpatPlugin/test/testPedestrianDetectionForSPAT.cpp

@@ -73,7 +73,7 @@ TEST(PedestrianDetectionForSPAT, updateEncodedSpat)
         EXPECT_EQ(spatPtr->intersections.list.array[0]->states.list.array[0]->signalGroup, 1);
         EXPECT_EQ(spatPtr->intersections.list.array[0]->states.list.array[1]->signalGroup, 2);
         ASSERT_NE(spatPtr->intersections.list.array[0]->maneuverAssistList, nullptr);
-        EXPECT_EQ(spatPtr->intersections.list.array[0]->maneuverAssistList->list.count, 1);
+        EXPECT_NE(spatPtr->intersections.list.array[0]->maneuverAssistList->list.count, 0);
         ASSERT_NE(spatPtr->intersections.list.array[0]->maneuverAssistList->list.array[0]->pedBicycleDetect, nullptr);
         EXPECT_EQ(*(spatPtr->intersections.list.array[0]->maneuverAssistList->list.array[0]->pedBicycleDetect), 1);
     }

src/v2i-hub/TelematicBridgePlugin/CMakeLists.txt

@@ -0,0 +1,19 @@
+PROJECT (TelematicBridgePlugin VERSION 7.5.1 LANGUAGES CXX)
+
+set (TMX_PLUGIN_NAME "Telematic Bridge")
+
+find_package(RapidJSON REQUIRED)
+
+BuildTmxPlugin()
+TARGET_LINK_LIBRARIES ( ${PROJECT_NAME} tmxutils jsoncpp)
+
+####################################################
+##################  Testing  #######################
+####################################################
+enable_testing()
+include_directories(${PROJECT_SOURCE_DIR}/src)
+add_library(${PROJECT_NAME}_lib src/TelematicBridgeMsgWorker.cpp)
+target_link_libraries(${PROJECT_NAME}_lib  PUBLIC tmxutils jsoncpp)
+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 ${PROJECT_NAME}_lib gtest)

src/v2i-hub/TelematicBridgePlugin/manifest.json

@@ -0,0 +1,16 @@
+{
+    "name": "TelematicBridge",
+    "description": "Plugin that listens for TMX messages and forward them to the Telematic cloud services.",
+    "version": "@PROJECT_VERSION@",
+    "exeLocation": "/bin/TelematicBridgePlugin",
+    "coreIpAddr": "127.0.0.1",
+    "corePort": 24601,
+    "messageTypes": [],
+    "configuration": [
+        {
+            "key": "LogLevel",
+            "default": "INFO",
+            "description": "The log level for this plugin"
+        }
+    ]
+}

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

@@ -0,0 +1,11 @@
+#pragma once
+#include <tmx/TmxException.hpp>
+
+namespace TelematicBridge
+{
+    class TelematicBridgeException : public tmx::TmxException
+    {
+    public:
+       using TmxException::TmxException;
+    };
+}

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

@@ -0,0 +1,184 @@
+#include "TelematicBridgeMsgWorker.h"
+
+namespace TelematicBridge
+{
+    bool TelematicBridgeMsgWorker::HexToBytes(const string &hexPaylod, vector<char> &byteBuffer)
+    {
+        uint8_t d = 0;
+        int i = 0;
+
+        for (const char &c : hexPaylod)
+        {
+            if (c <= '9' && c >= '0')
+            {
+                d = c - '0';
+            }
+            else if (c <= 'F' && c >= 'A')
+            {
+                d = c - 55; // c - 'A' + 10
+            }
+            else if (c <= 'f' && c >= 'a')
+            {
+                d = c - 87; // c - 'a' + 10;
+            }
+            else
+            {
+                return false;
+            }
+
+            if (i % 2)
+            {
+                // low order nibble.
+                byteBuffer.back() |= d;
+            }
+            else
+            {
+                // high order nibble.
+                byteBuffer.push_back(d << 4);
+            }
+            ++i;
+        }
+        return true;
+    }
+
+    void TelematicBridgeMsgWorker::DecodeJ2735Msg(const string &hexPaylod, MessageFrame_t *messageFrame)
+    {
+        /**
+         * Decode J2735 message
+         */
+        ostringstream erroross;
+        vector<char> byte_buffer;
+        if (!HexToBytes(hexPaylod, byte_buffer))
+        {
+            throw TelematicBridgeException("Failed attempt to decode MessageFrame hex string: cannot convert to bytes.");
+        }
+        asn_dec_rval_t decode_rval = asn_decode(
+            nullptr,
+            ATS_UNALIGNED_BASIC_PER,
+            &asn_DEF_MessageFrame,
+            (void **)&messageFrame,
+            byte_buffer.data(),
+            byte_buffer.size());
+
+        if (decode_rval.code != RC_OK)
+        {
+            erroross.str("");
+            erroross << "failed ASN.1 binary decoding of element " << asn_DEF_MessageFrame.name << ": bad data. Successfully decoded " << decode_rval.consumed << " bytes.";
+            throw TelematicBridgeException(erroross.str());
+        }
+    }
+
+    string TelematicBridgeMsgWorker::ConvertJ2735FrameToXML(const MessageFrame_t *messageFrame)
+    {
+        /**
+         * Convert J2735 message into XML
+         */
+        buffer_structure_t xml_buffer = {nullptr, 0, 0};
+        asn_enc_rval_t encode_rval = xer_encode(
+            &asn_DEF_MessageFrame,
+            messageFrame,
+            XER_F_CANONICAL,
+            DynamicBufferAppend,
+            static_cast<void *>(&xml_buffer));
+        if (encode_rval.encoded == -1)
+        {
+            throw TelematicBridgeException("Failed to  convert message with ID (=" + to_string(messageFrame->messageId) + ") to XML ");
+        }
+        return string(xml_buffer.buffer);
+    }
+
+    int TelematicBridgeMsgWorker::DynamicBufferAppend(const void *buffer, size_t size, void *app_key)
+    {
+        auto *xb = static_cast<buffer_structure_t *>(app_key);
+
+        while (xb->buffer_size + size + 1 > xb->allocated_size)
+        {
+            // increase size of buffer.
+            size_t new_size = 2 * (xb->allocated_size ? xb->allocated_size : 64);
+            auto new_buf = static_cast<char *>(MALLOC(new_size));
+            if (!new_buf)
+                return -1;
+            // move old to new.
+            memcpy(new_buf, xb->buffer, xb->buffer_size);
+
+            FREEMEM(xb->buffer);
+            xb->buffer = new_buf;
+            xb->allocated_size = new_size;
+        }
+
+        memcpy(xb->buffer + xb->buffer_size, buffer, size);
+        xb->buffer_size += size;
+        // null terminate the string.
+        xb->buffer[xb->buffer_size] = '\0';
+        return 0;
+    }
+
+    string TelematicBridgeMsgWorker::JsonToString(const Json::Value &json)
+    {
+        Json::FastWriter fasterWirter;
+        string jsonStr = fasterWirter.write(json);
+        boost::replace_all(jsonStr, "\\n", "");
+        boost::replace_all(jsonStr, "\n", "");
+        boost::replace_all(jsonStr, "\\t", "");
+        boost::replace_all(jsonStr, "\\", "");
+        return jsonStr;
+    }
+
+    Json::Value TelematicBridgeMsgWorker::StringToJson(const string &str)
+    {
+        Json::Value root;
+        Json::Reader reader;
+        bool parsingSuccessful = reader.parse(str, root);
+        if (!parsingSuccessful)
+        {
+            throw TelematicBridgeException("Error parsing the string");
+        }
+        return root;
+    }
+
+    Json::Value TelematicBridgeMsgWorker::IvpMessageToJson(const IvpMessage *msg)
+    {
+        Json::Value json;
+        if (msg->type)
+        {
+            json["type"] = msg->type;
+        }
+
+        if (msg->subtype)
+        {
+            json["subType"] = msg->subtype;
+        }
+
+        if (msg->dsrcMetadata)
+        {
+            json["channel"] = msg->dsrcMetadata->channel;
+            json["psid"] = msg->dsrcMetadata->psid;
+        }
+
+        if (msg->encoding)
+        {
+            json["encoding"] = msg->encoding;
+        }
+
+        if (msg->source)
+        {
+            json["source"] = msg->source;
+        }
+        json["sourceId"] = msg->sourceId;
+        json["flags"] = msg->flags;
+        json["timestamp"] = msg->timestamp;
+        if (msg->payload)
+        {
+            if (msg->payload->type == cJSON_Number)
+            {
+                json["payload"] = (msg->payload->valueint == 0 ? msg->payload->valuedouble : static_cast<double>(msg->payload->valueint));
+            }
+            else
+            {
+                json["payload"] = StringToJson(cJSON_Print(msg->payload));
+            }
+        }
+
+        return json;
+    }
+} // TelematicBridge