feature: establish connection to the endpoint by accessor when needed

rdkcentral · Dec 6, 2024 · a29a011 · a29a011
1 parent 7d65f54
commit a29a011
Show file tree

Hide file tree

Showing 2 changed files with 86 additions and 5 deletions.
diff --git a/languages/cpp/src/shared/src/Accessor/Accessor.cpp b/languages/cpp/src/shared/src/Accessor/Accessor.cpp
@@ -18,6 +18,8 @@
 
 #include "Accessor.h"
 
+#include <chrono>
+
 namespace FireboltSDK {
 
     Accessor* Accessor::_singleton = nullptr;
@@ -65,13 +67,16 @@ namespace FireboltSDK {
         return Event::Instance();
     }
 
-    Firebolt::Error Accessor::CreateTransport(const string& url, const Transport<WPEFramework::Core::JSON::IElement>::Listener& listener, const uint32_t waitTime = DefaultWaitTime)
+    Firebolt::Error Accessor::CreateTransport(const string& url, const uint32_t waitTime = DefaultWaitTime)
     {
         if (_transport != nullptr) {
             delete _transport;
         }
 
-        _transport = new Transport<WPEFramework::Core::JSON::IElement>(static_cast<WPEFramework::Core::URL>(url), waitTime, listener);
+        _transport = new Transport<WPEFramework::Core::JSON::IElement>(
+                static_cast<WPEFramework::Core::URL>(url),
+                waitTime,
+                std::bind(&Accessor::ConnectionChanged, this, std::placeholders::_1, std::placeholders::_2));
 
         ASSERT(_transport != nullptr);
         return ((_transport != nullptr) ? Firebolt::Error::None : Firebolt::Error::Timedout);
@@ -86,9 +91,38 @@ namespace FireboltSDK {
         return Firebolt::Error::None;
     }
 
+    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);
+        }
+    }
+
     Transport<WPEFramework::Core::JSON::IElement>* Accessor::GetTransport()
     {
-        ASSERT(_transport != nullptr);
+        if (_transport == nullptr || ! _connected) { // Try to connect if not connected: application has not yet connected or connection has been lost
+            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();
+            }
+        }
+
         return _transport;
     }
 

diff --git a/languages/cpp/src/shared/src/Accessor/Accessor.h b/languages/cpp/src/shared/src/Accessor/Accessor.h
@@ -25,6 +25,9 @@
 #include "Event/Event.h"
 #include "Logger/Logger.h"
 
+#include <condition_variable>
+#include <mutex>
+
 namespace FireboltSDK {
     class Accessor {
     private:
@@ -107,16 +110,30 @@ namespace FireboltSDK {
 
         Firebolt::Error Connect(const Transport<WPEFramework::Core::JSON::IElement>::Listener& listener)
         {
-            Firebolt::Error status = CreateTransport(_config.WsUrl.Value().c_str(), listener, _config.WaitTime.Value());
+            RegisterConnectionChangeListener(listener);
+            Firebolt::Error status = CreateTransport(_config.WsUrl.Value().c_str(), _config.WaitTime.Value());
             if (status == Firebolt::Error::None) {
                 Async::Instance().Configure(_transport);
                 status = CreateEventHandler();
             }
             return status;
         }
 
+        void RegisterConnectionChangeListener(const Transport<WPEFramework::Core::JSON::IElement>::Listener& listener)
+        {
+            _connectionChangeListener = listener;
+        }
+
+        void UnregisterConnnectionChangeListener()
+        {
+            _connectionChangeListener = nullptr;
+        }
+
         Firebolt::Error Disconnect()
         {
+            if (_transport == nullptr) {
+                return Firebolt::Error::None;
+            }
             Firebolt::Error status = Firebolt::Error::None;
             status = DestroyTransport();
             if (status == Firebolt::Error::None) {
@@ -126,19 +143,49 @@ namespace FireboltSDK {
             return status;
         }
 
+        bool IsConnected() const
+        {
+            return _connected;
+        }
+
         Event& GetEventManager();
         Transport<WPEFramework::Core::JSON::IElement>* GetTransport();
 
     private:
         Firebolt::Error CreateEventHandler();
         Firebolt::Error DestroyEventHandler();
-        Firebolt::Error CreateTransport(const string& url, const Transport<WPEFramework::Core::JSON::IElement>::Listener& listener, const uint32_t waitTime);
+        Firebolt::Error CreateTransport(const string& url, const uint32_t waitTime);
         Firebolt::Error DestroyTransport();
 
+        void ConnectionChanged(const bool connected, const Firebolt::Error error);
+
     private:
         WPEFramework::Core::ProxyType<WorkerPoolImplementation> _workerPool;
         Transport<WPEFramework::Core::JSON::IElement>* _transport;
         static Accessor* _singleton;
         Config _config;
+        struct {
+            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) {
+                std::unique_lock lk(m);
+                bool ret = cv.wait_for(lk, std::chrono::milliseconds(duration_ms), [&]{ return ready; });
+                lk.unlock();
+                return ret;
+            }
+            void signal() {
+                std::lock_guard lk(m);
+                ready = true;
+                cv.notify_one();
+            }
+        } _connectionChangeSync; // Synchronize a thread that is waiting for a connection if that one that is notified about connection changes
+
+        bool _connected = false;
+        Transport<WPEFramework::Core::JSON::IElement>::Listener _connectionChangeListener = nullptr;
     };
 }