Use rapidjson to parse headers, no more dict

Signed-off-by: Nijat K <[email protected]>

Author: NeejWeej

cpp/csp/adapters/websocket/ClientConnectionRequestAdapter.cpp

@@ -1,6 +1,4 @@
 #include <csp/adapters/websocket/ClientConnectionRequestAdapter.h>
-#include <csp/python/Conversions.h>
-#include <Python.h>
 
 namespace csp::adapters::websocket {
 
@@ -31,15 +29,35 @@ void ClientConnectionRequestAdapter::executeImpl()
     if (unlikely(m_isPruned))
         return;
 
-    auto raw_val = input()->lastValueTyped<PyObject*>();
-    auto val = python::fromPython<std::vector<Dictionary>>(raw_val);
+    std::vector<Dictionary> properties_list;
+    for (auto& request : input()->lastValueTyped<std::vector<InternalConnectionRequest::Ptr>>()) {
+        if (!request->allFieldsSet())
+            CSP_THROW(TypeError, "All fields must be set in InternalConnectionRequest");
+            
+        Dictionary dict;
+        dict.update("host", request->host());
+        dict.update("port", request->port());
+        dict.update("route", request->route());
+        dict.update("uri", request->uri());
+        dict.update("use_ssl", request->use_ssl());
+        dict.update("reconnect_interval", request->reconnect_interval());
+        dict.update("persistent", request->persistent());
+        
+        dict.update("headers", request -> headers() );
+        dict.update("on_connect_payload", request->on_connect_payload());
+        dict.update("action", request->action());
+        dict.update("dynamic", request->dynamic());
+        dict.update("binary", request->binary());
+        
+        properties_list.push_back(std::move(dict));
+    }
 
     // We intentionally post here, we want the thread running
     // the strand to handle the connection request. We want to keep
     // all updates to internal data structures at graph run-time
     // to that thread.
-    boost::asio::post(m_strand, [this, val=std::move(val)]() {
-        for(const auto& conn_req: val) {
+    boost::asio::post(m_strand, [this, properties_list=std::move(properties_list)]() {
+        for(const auto& conn_req: properties_list) {
             m_websocketManager->handleConnectionRequest(conn_req, m_callerId, m_isSubscribe);
         }
     });

cpp/csp/adapters/websocket/WebsocketEndpoint.cpp

@@ -1,14 +1,22 @@
 #include <csp/adapters/websocket/WebsocketEndpoint.h>
+#include <rapidjson/document.h>
 
 namespace csp::adapters::websocket {
 using namespace csp;
 
 WebsocketEndpoint::WebsocketEndpoint(
     net::io_context& ioc,
     Dictionary properties 
-) : m_properties(std::make_shared<Dictionary>(std::move(properties))),
+) : m_properties( std::make_shared<Dictionary>( std::move( properties ) ) ),
     m_ioc(ioc)
-{ };
+{
+    std::string headerProps = m_properties->get<std::string>("headers");
+    // Create new empty headers dictionary
+    auto headers = std::make_shared<Dictionary>();
+    m_properties->update("headers", headers);
+    // Update with any existing header properties
+    updateHeaders(headerProps);
+}
 void WebsocketEndpoint::setOnOpen(void_cb on_open)
 { m_on_open = std::move(on_open); }
 void WebsocketEndpoint::setOnFail(string_cb on_fail)
@@ -76,6 +84,25 @@ void WebsocketEndpoint::updateHeaders(csp::Dictionary properties){
     }
 }
 
+void WebsocketEndpoint::updateHeaders(const std::string& properties) {
+    if( properties.empty() )
+        return;
+    DictionaryPtr headers = m_properties->get<DictionaryPtr>("headers");
+    rapidjson::Document doc;
+    doc.Parse(properties.c_str());
+    if (doc.IsObject()) {
+        // Windows builds complained when doc.GetObject was called in the for loop
+        const auto& obj = doc.GetObject();
+        for (const auto& member : obj) {
+            if (member.value.IsString()) {
+                std::string key = member.name.GetString();
+                std::string value = member.value.GetString();
+                headers->update(key, std::move(value));
+            }
+        }
+    }
+}
+
 std::shared_ptr<Dictionary> WebsocketEndpoint::getProperties() {
     return m_properties;
 }

cpp/csp/adapters/websocket/WebsocketEndpoint.h

@@ -379,8 +379,8 @@ class WebsocketEndpoint {
     void setOnClose(void_cb on_close);
     void setOnSendFail(string_cb on_send_fail);
     void updateHeaders(Dictionary properties);
+    void updateHeaders(const std::string& properties);
     std::shared_ptr<Dictionary> getProperties();
-    // Dictionary& getProperties();
     void run();
     void stop( bool stop_ioc = true);
     void send(const std::string& s);

cpp/csp/adapters/websocket/WebsocketEndpointManager.cpp

@@ -241,12 +241,8 @@ void WebsocketEndpointManager::handleConnectionRequest(const Dictionary & proper
     // This should only get called from the thread running
     // m_ioc. This allows us to avoid locks on internal data
     // structures
-    // std::cout << " HEY YA\n";
     auto endpoint_id = properties.get<std::string>("uri");
-    // std::cout << endpoint_id;
     autogen::ActionType action = autogen::ActionType::create( properties.get<std::string>("action") );
-    // std::cout << action.asString() << "\n";
-    // // Change headers if needed here!
     switch(action.enum_value()) {
         case autogen::ActionType::enum_::CONNECT: {
             auto persistent = properties.get<bool>("persistent");
@@ -289,10 +285,9 @@ void WebsocketEndpointManager::handleConnectionRequest(const Dictionary & proper
                     m_endpoints[endpoint_id]->send(payload);
                 // Conscious decision to let non-persisten connection
                 // results to update the header
-                auto headers = *properties.get<DictionaryPtr>("headers");
+                auto headers = properties.get<std::string>("headers");
                 m_endpoints[endpoint_id]->updateHeaders(std::move(headers));
             }
-            // }
             break;
         }
 

csp/adapters/websocket.py

@@ -18,7 +18,13 @@
     RawBytesMessageMapper,
     RawTextMessageMapper,
 )
-from csp.adapters.websocket_types import ActionType, ConnectionRequest, WebsocketHeaderUpdate, WebsocketStatus
+from csp.adapters.websocket_types import (
+    ActionType,
+    ConnectionRequest,
+    InternalConnectionRequest,
+    WebsocketHeaderUpdate,
+    WebsocketStatus,
+)
 from csp.impl.wiring import input_adapter_def, output_adapter_def, status_adapter_def
 from csp.impl.wiring.delayed_node import DelayedNodeWrapperDef
 from csp.lib import _websocketadapterimpl
@@ -396,18 +402,6 @@ def _instantiate(self):
         _launch_application(self._port, manager, csp.const("stub"))
 
 
-# Maybe, we can have the Adapter manager have all the connections
-# If we get a new connection request, we include that adapter for the
-# subscriptions. When we pop it, we remove it.
-# Then, each edge will effectively be independent.
-# Maybe. have each websocket push to a shared queue, then from there we
-# pass it along to all edges ("input adapters") that are subscribed to it
-
-# Ok, maybe, let's keep it at just 1 subscribe and send call.
-# However, we can subscribe to the send and subscribe calls separately.
-# We just have to keep track of the Endpoints we have, and
-
-
 class WebsocketAdapterManager:
     """
     Can subscribe dynamically via ts[List[ConnectionRequest]]
@@ -453,7 +447,7 @@ def __init__(
                 connection_request = ConnectionRequest(
                     uri=uri, reconnect_interval=reconnect_interval, headers=headers or {}
                 )
-            self._properties.update(self._get_properties(connection_request))
+            self._properties.update(self._get_properties(connection_request).to_dict())
 
         # This is a counter that will be used to identify every function call
         # We keep track of the subscribes and sends separately
@@ -467,7 +461,7 @@ def __init__(
     def _dynamic(self):
         return self._properties.get("dynamic", False)
 
-    def _get_properties(self, conn_request: ConnectionRequest) -> dict:
+    def _get_properties(self, conn_request: ConnectionRequest) -> InternalConnectionRequest:
         uri = conn_request.uri
         reconnect_interval = conn_request.reconnect_interval
 
@@ -476,14 +470,14 @@ def _get_properties(self, conn_request: ConnectionRequest) -> dict:
         if resp.hostname is None:
             raise ValueError(f"Failed to parse host from URI: {uri}")
 
-        res = dict(
+        res = InternalConnectionRequest(
             host=resp.hostname,
             # if no port is explicitly present in the uri, the resp.port is None
             port=_sanitize_port(uri, resp.port),
             route=resp.path or "/",  # resource shouldn't be empty string
             use_ssl=uri.startswith("wss"),
             reconnect_interval=reconnect_interval,
-            headers=conn_request.headers,
+            headers=rapidjson.dumps(conn_request.headers) if conn_request.headers else "",
             persistent=conn_request.persistent,
             action=conn_request.action.name,
             on_connect_payload=conn_request.on_connect_payload,
@@ -537,7 +531,9 @@ def subscribe(
         adapter_props = AdapterInfo(caller_id=caller_id, is_subscribe=True).model_dump()
         connection_request = csp.null_ts(List[ConnectionRequest]) if connection_request is None else connection_request
         request_dict = csp.apply(
-            connection_request, lambda conn_reqs: [self._get_properties(conn_req) for conn_req in conn_reqs], list
+            connection_request,
+            lambda conn_reqs: [self._get_properties(conn_req) for conn_req in conn_reqs],
+            List[InternalConnectionRequest],
         )
         # Output adapter to handle connection requests
         _websocket_connection_request_adapter_def(self, request_dict, adapter_props)
@@ -566,7 +562,9 @@ def send(self, x: ts["T"], connection_request: Optional[ts[List[ConnectionReques
         adapter_props = AdapterInfo(caller_id=caller_id, is_subscribe=False).model_dump()
         connection_request = csp.null_ts(List[ConnectionRequest]) if connection_request is None else connection_request
         request_dict = csp.apply(
-            connection_request, lambda conn_reqs: [self._get_properties(conn_req) for conn_req in conn_reqs], list
+            connection_request,
+            lambda conn_reqs: [self._get_properties(conn_req) for conn_req in conn_reqs],
+            List[InternalConnectionRequest],
         )
         _websocket_connection_request_adapter_def(self, request_dict, adapter_props)
         return _websocket_output_adapter_def(self, x, adapter_props)
@@ -614,6 +612,6 @@ def _create(self, engine, memo):
     "websocket_connection_request_adapter",
     _websocketadapterimpl._websocket_connection_request_adapter,
     WebsocketAdapterManager,
-    input=ts[list],  # needed, List[dict] didn't work on c++ level
+    input=ts[List[InternalConnectionRequest]],  # needed, List[dict] didn't work on c++ level
     properties=dict,
 )

csp/adapters/websocket_types.py

@@ -34,3 +34,25 @@ class ConnectionRequest(Struct):
     reconnect_interval: timedelta = timedelta(seconds=2)
     on_connect_payload: str = ""  # message to send on connect
     headers: Dict[str, str] = {}
+
+
+# Only used internally
+class InternalConnectionRequest(Struct):
+    host: str  # Hostname parsed from the URI
+    port: str  # Port number for the connection (parsed and sanitized from URI)
+    route: str  # Resource path from URI, defaults to "/" if empty
+    uri: str  # Complete original URI string
+
+    # Connection behavior
+    use_ssl: bool  # Whether to use secure WebSocket (wss://)
+    reconnect_interval: timedelta  # Time to wait between reconnection attempts
+    persistent: bool  # Whether to maintain a persistent connection
+
+    # Headers and payloads
+    headers: str  # HTTP headers for the connection as json string
+    on_connect_payload: str  # Message to send when connection is established
+
+    # Connection metadata
+    action: str  # Connection action type (Connect, Disconnect, Ping, etc)
+    dynamic: bool  # Whether the connection is dynamic
+    binary: bool  # Whether to use binary mode for the connection