Merge branch 'develop' into 549-move-time-kafka-producer

docs/Release_notes.md

@@ -1,6 +1,44 @@
 V2X-Hub Release Notes
 ---------------------------------
 
+Version 7.5.1, released June 21st, 2023
+--------------------------------------------------------
+
+**Summary:**
+ V2X Hub release 7.5.1 includes added functionality to integrate V2X Hub with CDASim environment. This integration includes V2X Hub registering as a Roadside Unit (RSU) in the CDASim environment, consuming and producing J2735 messages to the CDASim environment, and adding functionality to synchronize plugins to CDASim simulation time.
+
+**<ins>V2X Hub CDASim Functionalities </ins>**
+
+Enhancements in this release: 
+
+- Added new carma-time-lib to V2X Hub to allow services to use an external source for time value and update rate. 
+- Added new Plugin Client ClockAware, which extends Plugin Client and implements functionality to consume time sync messages and updates a carma-clock object from carma-time-lib. Any plugins that want to synchronize their time to simulation must extend this plugin to gain access to this functionality. 
+- Added CDASim Adapter plugin which is responsible for establishing connection between V2X Hub and CDASim environment. This includes a handshake that provides information about the V2X Hub simulated location and ID and message forwarding for J2735 messages and time synchronization messages. This plugin requires several environment variables to be set which are documented on the GitHub repo README.md. 
+
+Fixes in this release: 
+
+- PR 488: Added a simulated clock functionality with the new time library and tested. 
+- PR 489: Setup Kafka consumers for the time topic when running in simulation mode. 
+- Issue 492: Created a carma-simulation adapter shell for service that will act as adapter for CARMA Simulation integration. 
+- PR 509: Added a V2X Hub plugin inside the simulation platform to receive all messages from V2X Hub. This plugin contains parameters and variables that are provided in real-world scenarios. 
+- Issue 514: Added handshake functionality to carma-simulation ambassador instance which register’s the V2X Hub instance inside the simulator to allow multiple V2X Hub instances to connect with a single CARMA Simulation platform. 
+- Issue 535: Updated infrastructure registration to use a cartesian point as location over a geodetic point to allow for easier configuration of simulated location of an RSU. 
+- Issue 537: Fixed configuration parameters to correctly map X, Y, Z coordinates to Point for Infrastructure registration in CDASim Adapter. 
+- Issue 525: Fixed CDASim Adapter plugin that throws an exception while attempting CDASim handshake with CARMA-Simulation. 
+
+Known issues in this release: 
+
+- Issue #540: CDASim Time Synchronization is non-time-regulating. If simulation runs too fast (faster than real-time) for V2X Hub to keep up, V2X Hub can fall behind in time. 
+- Issue #507: SPaT plugin throws segfault when in SIM MODE  
+- Issue #10 in carma-time-lib (not V2X Hub repo): wait_for_initialization does not support notifying multiple threads Work around exists for services/plugins using carma-time-lib. 
+- Issue #543: CARMA Streets Plugin Kafka Consumers can send redundant subscription attempts on initialization and can cause subscriptions to silently fail. 
+
+**<ins>Other </ins>**
+
+Enhancements in this release: 
+
+- Issue 511: Added new functionality get the log time for a message received in V2xHub to forward to Carma cloud, and from receiving in Carma Cloud to forward V2xhub. 
+
 Version 7.5.0, released May 5th, 2023
 --------------------------------------------------------
 

src/tmx/TmxUtils/src/Point.h

@@ -0,0 +1,19 @@
+#pragma once
+
+namespace tmx::utils {
+
+
+    /// Cartesian Coordinates on a .
+    typedef struct Point
+    {
+        Point() : X(0), Y(0), Z(0) {}
+
+        Point(double x, double y, double z = 0.0):
+            X(x), Y(y), Z(z) { }
+
+        double X;
+        double Y;
+        double Z;
+    } Point;
+
+} // namespace tmx::utils

src/v2i-hub/CARMACloudPlugin/CMakeLists.txt

@@ -1,4 +1,4 @@
-PROJECT ( CARMACloudPlugin VERSION 7.5.0 LANGUAGES CXX )
+PROJECT ( CARMACloudPlugin VERSION 7.5.1 LANGUAGES CXX )
 
 SET (TMX_PLUGIN_NAME "CARMACloud")
 add_compile_options(-fPIC)

src/v2i-hub/CARMAStreetsPlugin/CMakeLists.txt

@@ -1,4 +1,4 @@
-PROJECT ( CARMAStreetsPlugin VERSION 7.5.0 LANGUAGES CXX )
+PROJECT ( CARMAStreetsPlugin VERSION 7.5.1 LANGUAGES CXX )
 
 BuildTmxPlugin ( )
 

src/v2i-hub/CDASimAdapter/CMakeLists.txt

@@ -1,4 +1,4 @@
-PROJECT ( CDASimAdapter VERSION 7.5.0 LANGUAGES CXX )
+PROJECT ( CDASimAdapter VERSION 7.5.1 LANGUAGES CXX )
 set(CMAKE_CXX_STANDARD 17)
 
 FIND_PACKAGE( carma-clock )

src/v2i-hub/CDASimAdapter/manifest.json

@@ -14,20 +14,31 @@
             "description":"The log level for this plugin"
         },
 		{
-			"key":"Longitude",
+			"key":"X",
 			"default":"0.0",
-			"description":"Longitude (in degrees) of the location of the simulated infrastructure."
+			"description":"Cartesian X coordinate in simulated map (in meters)."
 		},
 		{
-			"key":"Latitude",
+			"key":"Y",
 			"default":"0.0",
-			"description":"Latitude (in degrees) of the location of the simulated infrastructure."
+			"description":"Cartesian Y coordinate in simulated map (in meters)."
 		},
 		{
-			"key":"Elevation",
+			"key":"Z",
 			"default":"0.0",
-			"description":"Elevation (in degrees) of the location of the simulated infrastructure."
+			"description":"Cartesian Z coordinate in simulated map (in meters)."
+		},
+		{
+			"key":"MaxConnectionAttempts",
+			"default":"10",
+			"description":"Number of connection attempts CDASimAdapter plugin will execute before failing. Valid values are any integers greater than or equal to 1. Any value less than 1 will result in unlimited connection attemtps."
+		},
+		{
+			"key":"ConnectionSleepTime",
+			"default":"4",
+			"description":"Number of seconds to wait after a failed CDASim connection attempt, before retrying to establish connection. Valid values are equal or greater than 1. Any value less than 1 will be treated as 1."
 		}
+
 
 	]
 }

src/v2i-hub/CDASimAdapter/src/CDASimAdapter.cpp

@@ -13,11 +13,22 @@ namespace CDASimAdapter{
     }
 
     void CDASimAdapter::UpdateConfigSettings() {
-        GetConfigValue<double>("Longitude", location.Longitude);
-        GetConfigValue<double>("Latitude", location.Latitude);
-        GetConfigValue<double>("Elevation", location.Elevation);
-        PLOG(logINFO) << "Location of Simulated V2X-Hub updated to : {" << location.Longitude << ", " 
-            << location.Latitude << ", " << location.Elevation << "}." << std::endl;
+        std::scoped_lock<std::mutex> lock(_lock);
+        bool success = false;
+        success = GetConfigValue<double>("X", location.X);
+        success = success && GetConfigValue<double>("Y", location.Y);
+        success = success && GetConfigValue<double>("Z", location.Z);
+        PLOG(logINFO) << "Location of Simulated V2X-Hub updated to : {" << location.X << ", " 
+            << location.Y << ", " << location.Z << "}." << std::endl;
+        success = success && GetConfigValue<int>("MaxConnectionAttempts", max_connection_attempts);
+        success = success && GetConfigValue<uint>("ConnectionSleepTime", connection_sleep_time);
+        if (connection_sleep_time < 1 ) {
+            PLOG(logWARNING) << "ConnectionSleepTime of " << connection_sleep_time << " is invalid. Valid values are <= 1." << std::endl;      
+            connection_sleep_time = 1;
+        }
+        if (!success) {
+            PLOG(logWARNING) << "Some configuration parameters were not successfully loaded! Please ensure configuration parameter keys are correct!" << std::endl;
+        }
     }
 
     void CDASimAdapter::OnConfigChanged(const char *key, const char *value) {
@@ -32,14 +43,27 @@ namespace CDASimAdapter{
             UpdateConfigSettings();
 
             // While CARMA Simulation connection is down, attempt to reconnect
-            while ( !connection || !connection->is_connected() ) {
-                connect();
+            int connection_attempts = 0;
+            while ( (!connection || !connection->is_connected()) && (connection_attempts < max_connection_attempts || max_connection_attempts < 1 ) ) {
+                PLOG(logINFO) << "Attempting CDASim connection " << connection_attempts << "/" << max_connection_attempts << " ..." << std::endl;
+                bool success = connect();
+                if (success) {
+                    PLOG(logINFO) << "Connection to CDASim established!" << std::endl;
+                }
+                connection_attempts++;
+                // Sleep for configurable seconds in between connection attempts. No sleep is required on final failing attempt
+                if ( !connection->is_connected() && (connection_attempts < max_connection_attempts || max_connection_attempts < 1 ) ) {
+                    PLOG(logDEBUG) << "Sleeping for " << connection_sleep_time << " seconds before next connection attempt ..." << std::endl;
+                    std::this_thread::sleep_for(std::chrono::seconds(connection_sleep_time));
+                }
             }
 
             if ( connection->is_connected() ) {
                 start_time_sync_thread_timer();
-                start_amf_msg_thread();
-                start_binary_msg_thread();
+                start_immediate_forward_thread();
+                start_message_receiver_thread();
+            }else {
+                PLOG(logERROR) << "CDASim connection failed!" << std::endl;
             }
 
         }
@@ -63,7 +87,7 @@ namespace CDASimAdapter{
             uint time_sync_port = std::stoul(sim::get_sim_config(sim::TIME_SYNC_PORT));
             uint v2x_port = std::stoul(sim::get_sim_config(sim::V2X_PORT));
             uint sim_v2x_port = std::stoul(sim::get_sim_config(sim::SIM_V2X_PORT));
-            uint infrastructure_id = std::stoul(sim::get_sim_config(sim::INFRASTRUCTURE_ID));;
+            std::string infrastructure_id = sim::get_sim_config(sim::INFRASTRUCTURE_ID);
 
             PLOG(logINFO) << "CDASim connecting " << simulation_ip << 
                     "\nUsing Registration Port : "  << std::to_string( simulation_registration_port) <<
@@ -80,36 +104,42 @@ namespace CDASimAdapter{
         catch (const TmxException &e) {
             PLOG(logERROR) << "Exception occured attempting to initialize CDASim Connection : " << e.what() << std::endl;
             return false;
+        }
+        catch (const std::invalid_argument &e ) {
+            // std::stoul throws invalid arguement exception when provided with a string that contains characters that are not numbers.
+            PLOG(logERROR) << "Exception occured attempting to initialize CDASim Connection : " << e.what() << 
+                ". Check environment variables are set to the correct type!";
+            return false;
         }   
         return connection->connect();
     }
 
 
 
-    void CDASimAdapter::start_amf_msg_thread() {
-        if ( !amf_thread_timer ) {
-            amf_thread_timer = std::make_unique<tmx::utils::ThreadTimer>();
+    void CDASimAdapter::start_immediate_forward_thread() {
+        if ( !immediate_forward_timer ) {
+            immediate_forward_timer = std::make_unique<tmx::utils::ThreadTimer>();
         }
-        amf_msg_tick_id = amf_thread_timer->AddPeriodicTick([this]() {
+        immediate_forward_tick_id = immediate_forward_timer->AddPeriodicTick([this]() {
             this->attempt_message_from_v2xhub();
 
         } // end of lambda expression
         , std::chrono::milliseconds(100) );
 
-        amf_thread_timer->Start();
+        immediate_forward_timer->Start();
 
     }
 
-    void CDASimAdapter::start_binary_msg_thread() {
-        if ( !binary_thread_timer ) {
-            binary_thread_timer = std::make_unique<tmx::utils::ThreadTimer>();
+    void CDASimAdapter::start_message_receiver_thread() {
+        if ( !message_receiver_timer ) {
+            message_receiver_timer = std::make_unique<tmx::utils::ThreadTimer>();
         }
 
-        binary_msg_tick_id = binary_thread_timer->AddPeriodicTick([this]() {
+        message_receiver_tick_id = message_receiver_timer->AddPeriodicTick([this]() {
             this->attempt_message_from_simulation();
         } // end of lambda expression
         , std::chrono::milliseconds(100) );
-        binary_thread_timer->Start();
+        message_receiver_timer->Start();
 
     }
 
@@ -149,15 +179,15 @@ namespace CDASimAdapter{
 
     void CDASimAdapter::start_time_sync_thread_timer() {
         PLOG(logDEBUG) << "Creating Thread Timer for time sync" << std::endl;
-        if ( !thread_timer ) {
-            thread_timer = std::make_unique<tmx::utils::ThreadTimer>();
+        if ( !time_sync_timer ) {
+            time_sync_timer = std::make_unique<tmx::utils::ThreadTimer>();
         }
-        time_sync_tick_id = thread_timer->AddPeriodicTick([this]() {
+        time_sync_tick_id = time_sync_timer->AddPeriodicTick([this]() {
             PLOG(logDEBUG1) << "Listening for time sync messages from CDASim." << std::endl;
             this->attempt_time_sync();
         } // end of lambda expression
         , std::chrono::milliseconds(100));
-        thread_timer->Start();
+        time_sync_timer->Start();
     }
 
     void CDASimAdapter::attempt_time_sync() {

src/v2i-hub/CDASimAdapter/src/CDASimConnection.cpp

@@ -4,9 +4,9 @@
 using namespace tmx::utils;
 
 namespace CDASimAdapter{ 
-    CDASimConnection::CDASimConnection(const std::string &simulation_ip, const uint infrastructure_id, const uint simulation_registration_port, const uint sim_v2x_port,
+    CDASimConnection::CDASimConnection(const std::string &simulation_ip, const std::string &infrastructure_id, const uint simulation_registration_port, const uint sim_v2x_port,
                                                         const std::string &local_ip,  const uint time_sync_port, const uint v2x_port, 
-                                                        const WGS84Point &location) : 
+                                                        const Point &location) : 
                                                         _simulation_ip(simulation_ip), _infrastructure_id(infrastructure_id), _simulation_registration_port(simulation_registration_port),
                                                         _simulation_v2x_port(sim_v2x_port), _local_ip(local_ip), _time_sync_port(time_sync_port), _v2x_port(v2x_port),
                                                         _location(location)  {
@@ -33,8 +33,8 @@ namespace CDASimAdapter{
         return _connected;
     }
 
-    std::string CDASimConnection::get_handshake_json(const uint infrastructure_id, const std::string &local_ip,  const uint time_sync_port, const uint v2x_port, 
-                                const WGS84Point &location) const
+    std::string CDASimConnection::get_handshake_json(const std::string &infrastructure_id, const std::string &local_ip,  const uint time_sync_port, const uint v2x_port, 
+                                const Point &location) const
 
     {
         Json::Value message;   
@@ -44,17 +44,17 @@ namespace CDASimAdapter{
         message["infrastructureId"] = infrastructure_id;
         message["rxMessagePort"] = v2x_port;
         message["timeSyncPort"] = time_sync_port;
-        message["location"]["latitude"] = location.Latitude;
-        message["location"]["longitude"] = location.Longitude;
-        message["location"]["elevation"] = location.Elevation;
+        message["location"]["x"] = location.X;
+        message["location"]["y"] = location.Y;
+        message["location"]["z"] = location.Z;
         Json::StyledWriter writer;
         message_str = writer.write(message);
         return message_str;
     }
 
-    bool CDASimConnection::carma_simulation_handshake(const std::string &simulation_ip, const uint infrastructure_id, const uint simulation_registration_port, 
+    bool CDASimConnection::carma_simulation_handshake(const std::string &simulation_ip, const std::string &infrastructure_id, const uint simulation_registration_port, 
                                 const std::string &local_ip,  const uint time_sync_port, const uint v2x_port, 
-                                const WGS84Point &location) 
+                                const Point &location) 
     {
         // Create JSON message with the content 
         std::string payload = "";