Bi-Directional gateway

languages/cpp/src/shared/CMakeLists.txt

@@ -24,9 +24,20 @@ set(FIREBOLT_LOGLEVEL "Info" CACHE STRING  "Log level to be enabled")
 
 # Default options
 option(FIREBOLT_ENABLE_STATIC_LIB "Create Firebolt library as Static library" OFF)
+option(ENABLE_BIDRECTIONAL "Enable bidirectional communication over WS" OFF)
 option(ENABLE_TESTS "Build openrpc native test" ON)
 option(ENABLE_UNIT_TESTS "Enable unit test" ON)
 option(ENABLE_COVERAGE "Enable code coverage build." ON)
+option(FIREBOLT_PLAIN_LOG "Disable log coloring" OFF)
+
+if (ENABLE_BIDRECTIONAL)
+    add_compile_definitions(GATEWAY_BIDIRECTIONAL)
+endif ()
+
+if (FIREBOLT_PLAIN_LOG)
+    add_compile_definitions(LOGGER_NO_COLOR)
+endif ()
+
 
 if (FIREBOLT_ENABLE_STATIC_LIB)
     set(FIREBOLT_LIBRARY_TYPE STATIC)
@@ -117,4 +128,4 @@ endif()
 # make sure others can make use cmake settings of Firebolt OpenRPC
 configure_file( "${CMAKE_SOURCE_DIR}/cmake/project.cmake.in"
         "${CMAKE_BINARY_DIR}/${FIREBOLT_NAMESPACE}Config.cmake"
-    @ONLY)
+    @ONLY)

languages/cpp/src/shared/include/json_engine.h

@@ -1,3 +1,22 @@
+/*
+ * Copyright 2024 Comcast Cable Communications Management, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+#pragma once
+
 #include<iostream>
 #include <fstream>
 

languages/cpp/src/shared/src/Accessor/Accessor.cpp

@@ -93,37 +93,35 @@ namespace FireboltSDK {
 
     void Accessor::ConnectionChanged(const bool connected, const Firebolt::Error error)
     {
-        _connectionChangeSync.signal(); // Signal waiting thread that the connection changed
         _connected = connected;
         if (_connectionChangeListener != nullptr) { // Notify a listener about the connection change
              _connectionChangeListener(connected, error);
         }
+        if (!_connected) {
+            _connectionChangeSync.signal(); // Signal to reconnect
+        }
     }
 
-    Transport<WPEFramework::Core::JSON::IElement>* Accessor::GetTransport()
+    void Accessor::Reconnector()
     {
-        if (_transport == nullptr || ! _connected) { // Try to connect if not connected: application has not yet connected or connection has been lost
+        while (running) {
+            _connectionChangeSync.wait(); // Wait for the signal that the connection has changed
+            if (!running) {
+                break;
+            }
+
+            Gateway::Instance().TransportUpdated(nullptr);
             DestroyTransport(); // Clean the transport if necessary
 
-            _connectionChangeSync.reset();
             Firebolt::Error status = CreateTransport( // Recreate the transport with the configuration passed to CTor
                 _config.WsUrl.Value().c_str(),
                 _config.WaitTime.Value());
-
-            bool ret = _connectionChangeSync.wait_for(_config.WaitTime.Value()); // Wait for the signal that the connection has changed, but no more than `WaitTime`
-
-            if (ret) { // Check if the connection successfully established
-                ASSERT(_transport != nullptr);
-                if (status == Firebolt::Error::None) { // If yes, proceed with the configuration of Async and Event-Handler
-                    Async::Instance().Configure(_transport);
-                    status = CreateEventHandler();
-                }
-            } else { // If the connection cannot be established, clean the transport
-                DestroyTransport();
+            if (status == Firebolt::Error::None) { // Proceed with the configuration of Async and Event-Handler
+                Async::Instance().Configure(_transport);
+                Gateway::Instance().TransportUpdated(_transport);
+                status = CreateEventHandler();
             }
         }
-
-        return _transport;
     }
 
 }

languages/cpp/src/shared/src/Accessor/Accessor.h

@@ -23,6 +23,7 @@
 #include "Transport/Transport.h"
 #include "Async/Async.h"
 #include "Event/Event.h"
+#include "Gateway/Gateway.h"
 #include "Logger/Logger.h"
 
 #include <condition_variable>
@@ -114,8 +115,11 @@ namespace FireboltSDK {
             Firebolt::Error status = CreateTransport(_config.WsUrl.Value().c_str(), _config.WaitTime.Value());
             if (status == Firebolt::Error::None) {
                 Async::Instance().Configure(_transport);
+                Gateway::Instance().TransportUpdated(_transport);
                 status = CreateEventHandler();
             }
+            running = true;
+            reconnector = std::thread(std::bind(&Accessor::Reconnector, this));
             return status;
         }
 
@@ -131,16 +135,12 @@ namespace FireboltSDK {
 
         Firebolt::Error Disconnect()
         {
-            if (_transport == nullptr) {
-                return Firebolt::Error::None;
+            running = false;
+            _connectionChangeSync.signal(); // Signal to reconnect
+            if (reconnector.joinable()) {
+                reconnector.join();
             }
-            Firebolt::Error status = Firebolt::Error::None;
-            status = DestroyTransport();
-            if (status == Firebolt::Error::None) {
-                Async::Dispose();
-                status = DestroyEventHandler();
-            }
-            return status;
+            return Firebolt::Error::None;
         }
 
         bool IsConnected() const
@@ -149,7 +149,6 @@ namespace FireboltSDK {
         }
 
         Event& GetEventManager();
-        Transport<WPEFramework::Core::JSON::IElement>* GetTransport();
 
     private:
         Firebolt::Error CreateEventHandler();
@@ -158,6 +157,7 @@ namespace FireboltSDK {
         Firebolt::Error DestroyTransport();
 
         void ConnectionChanged(const bool connected, const Firebolt::Error error);
+        void Reconnector();
 
     private:
         WPEFramework::Core::ProxyType<WorkerPoolImplementation> _workerPool;
@@ -168,15 +168,11 @@ namespace FireboltSDK {
             std::mutex m;
             std::condition_variable cv;
             bool ready = false;
-            void reset() {
-                std::lock_guard lk(m);
-                ready = false;
-            }
-            bool wait_for(unsigned duration_ms) {
+            void wait() {
                 std::unique_lock lk(m);
-                bool ret = cv.wait_for(lk, std::chrono::milliseconds(duration_ms), [&]{ return ready; });
+                cv.wait(lk, [&]{ return ready; });
                 lk.unlock();
-                return ret;
+                ready = false;
             }
             void signal() {
                 std::lock_guard lk(m);
@@ -186,6 +182,8 @@ namespace FireboltSDK {
         } _connectionChangeSync; // Synchronize a thread that is waiting for a connection if that one that is notified about connection changes
 
         bool _connected = false;
+        std::atomic<bool> running { false };
+        std::thread reconnector;
         Transport<WPEFramework::Core::JSON::IElement>::Listener _connectionChangeListener = nullptr;
     };
 }

languages/cpp/src/shared/src/CMakeLists.txt

@@ -22,15 +22,22 @@ set(TARGET ${PROJECT_NAME})
 message("Setup ${TARGET} v${PROJECT_VERSION}")
 
 file(GLOB SOURCES *.cpp)
-add_library(${TARGET} ${FIREBOLT_LIBRARY_TYPE}
-    ${SOURCES}
+list(APPEND SOURCES
     Logger/Logger.cpp
+    Gateway/Gateway.cpp
     Transport/Transport.cpp
     Accessor/Accessor.cpp
-    Event/Event.cpp
     Async/Async.cpp
 )
 
+if (ENABLE_BIDRECTIONAL)
+    list(APPEND SOURCES Event/bidi/Event.cpp)
+else ()
+    list(APPEND SOURCES Event/unidi/Event.cpp)
+endif ()
+
+add_library(${TARGET} ${FIREBOLT_LIBRARY_TYPE} ${SOURCES})
+
 if(ENABLE_UNIT_TESTS)
     target_compile_definitions(FireboltSDK PRIVATE UNIT_TEST)
 endif()
@@ -82,4 +89,4 @@ install(
 
 InstallHeaders(TARGET ${TARGET} HEADERS . NAMESPACE ${FIREBOLT_NAMESPACE} DESTINATION ${FIREBOLT_NAMESPACE}SDK)
 InstallCMakeConfig(TARGETS ${TARGET})
-InstallPackageConfig(TARGETS ${TARGET} DESCRIPTION "Firebolt SDK Library")
+InstallPackageConfig(TARGETS ${TARGET} DESCRIPTION "Firebolt SDK Library")