adds burst to websocket adapter; move build to c++20

Signed-off-by: Will Rieger <will.r.rieger@gmail.com>
Point72 · Jul 20, 2024 · 1ef02d2 · 1ef02d2
1 parent cf7e633
commit 1ef02d2
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 8 deletions.
diff --git a/cpp/csp/adapters/websocket/ClientInputAdapter.cpp b/cpp/csp/adapters/websocket/ClientInputAdapter.cpp
@@ -28,11 +28,11 @@ ClientInputAdapter::ClientInputAdapter(
 void ClientInputAdapter::processMessage( void* c, size_t t, PushBatch* batch ) 
 {
 
-    if( type() -> type() == CspType::Type::STRUCT )
+    if( dataType() -> type() == CspType::Type::STRUCT )
     {
         auto tick = m_converter -> asStruct( c, t );
         pushTick( std::move(tick), batch );
-    } else if ( type() -> type() == CspType::Type::STRING )
+    } else if ( dataType() -> type() == CspType::Type::STRING )
     {
         pushTick( std::string((char const*)c, t), batch );
     }

diff --git a/cpp/csp/python/adapters/CMakeLists.txt b/cpp/csp/python/adapters/CMakeLists.txt
@@ -40,9 +40,7 @@ if(CSP_BUILD_PARQUET_ADAPTER)
 endif()
 
 if(CSP_BUILD_WS_CLIENT_ADAPTER)
-    set(CMAKE_CXX_STANDARD 17)
-        add_library(websocketadapterimpl SHARED websocketadapterimpl.cpp)
-      	target_link_libraries(websocketadapterimpl csp_core csp_engine cspimpl csp_websocket_client_adapter)
-      	install(TARGETS websocketadapterimpl RUNTIME DESTINATION ${CSP_RUNTIME_INSTALL_SUBDIR})
-    set(CMAKE_CXX_STANDARD 20)
+    add_library(websocketadapterimpl SHARED websocketadapterimpl.cpp)
+    target_link_libraries(websocketadapterimpl csp_core csp_engine cspimpl csp_websocket_client_adapter)
+    install(TARGETS websocketadapterimpl RUNTIME DESTINATION ${CSP_RUNTIME_INSTALL_SUBDIR})
 endif()
diff --git a/csp/adapters/websocket.py b/csp/adapters/websocket.py
@@ -443,7 +443,10 @@ def subscribe(
         properties["field_map"] = field_map
         properties["meta_field_map"] = meta_field_map
 
-        return _websocket_input_adapter_def(self, ts_type, properties, push_mode)
+        edge = _websocket_input_adapter_def(self, ts_type, properties, push_mode)
+        if push_mode == csp.PushMode.BURST:
+            edge.tstype = csp.ts[List[ts_type]]
+        return edge
 
     def send(self, x: ts["T"]):
         return _websocket_output_adapter_def(self, x)

diff --git a/csp/tests/adapters/test_websocket.py b/csp/tests/adapters/test_websocket.py
@@ -126,3 +126,30 @@ def g():
             csp.stop_engine(ws.status())
 
         csp.run(g, starttime=datetime.now(pytz.UTC), realtime=True)
+
+    def test_send_recv_burst_json(self):
+        class MsgStruct(csp.Struct):
+            a: int
+            b: str
+
+        @csp.node
+        def send_msg_on_open(status: ts[Status]) -> ts[str]:
+            if csp.ticked(status):
+                return MsgStruct(a=1234, b="im a string").to_json()
+
+        @csp.graph
+        def g():
+            ws = WebsocketAdapterManager("ws://localhost:8000/")
+            status = ws.status()
+            ws.send(send_msg_on_open(status))
+            recv = ws.subscribe(MsgStruct, JSONTextMessageMapper(), push_mode=csp.PushMode.BURST)
+
+            csp.add_graph_output("recv", recv)
+            csp.stop_engine(recv)
+
+        msgs = csp.run(g, starttime=datetime.now(pytz.UTC), realtime=True)
+        obj = msgs["recv"][0][1]
+        assert isinstance(obj, list)
+        innerObj = obj[0]
+        assert innerObj.a == 1234
+        assert innerObj.b == "im a string"