Remove print statements, add more to websocket tests

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

Author: Nijat K

cpp/csp/adapters/websocket/ClientAdapterManager.cpp

@@ -537,7 +537,6 @@ void ClientAdapterManager::stop() {
 
 PushInputAdapter* ClientAdapterManager::getInputAdapter(CspTypePtr & type, PushMode pushMode, const Dictionary & properties)
 {   
-    std::cout << "Adapter manage m_dynamic " << m_dynamic << "\n";
     if ( m_dynamic ){
         auto caller_id = properties.get<int64_t>("caller_id");
         size_t validated_id = validateCallerId(caller_id);

cpp/csp/adapters/websocket/ClientInputAdapter.cpp

@@ -52,7 +52,6 @@ void ClientInputAdapter::processMessage( std::tuple<std::string, void*> data, si
 {
     // Extract the source string and data pointer from tuple
     std::string source = std::get<0>(data);
-    std::cout << source << "\n";
     void* c = std::get<1>(data);
     // std::cout << "Got data pointer: " << c << "\n";
     // std::cout << "dynamic??\n";
@@ -78,7 +77,6 @@ void ClientInputAdapter::processMessage( std::tuple<std::string, void*> data, si
             auto msg =  std::string((char const*)c, t);
             // std::cout << "got message " << msg << "\n";
             actual_type.meta()->field("msg")->setValue( true_val.get(), msg );
-            std::cout << msg << "\n";
 
             pushTick( std::move(true_val), batch );
         }

csp/adapters/websocket.py

@@ -21,8 +21,7 @@
 from csp.impl.wiring.delayed_node import DelayedNodeWrapperDef
 from csp.lib import _websocketadapterimpl
 
-from .websocket_types import ActionType  # noqa
-from .websocket_types import ConnectionRequest, WebsocketHeaderUpdate
+from .websocket_types import ActionType, ConnectionRequest, WebsocketHeaderUpdate, WebsocketStatus  # noqa
 
 # InternalConnectionRequest,
 _ = (

csp/tests/adapters/test_websocket.py

@@ -23,6 +23,7 @@
         RawTextMessageMapper,
         Status,
         WebsocketAdapterManager,
+        WebsocketStatus,
     )
 
     class EchoWebsocketHandler(tornado.websocket.WebSocketHandler):
@@ -263,34 +264,52 @@ def g():
             csp.run(g, starttime=datetime.now(pytz.UTC), endtime=timedelta(seconds=1), realtime=True)
 
     def test_dynamic_multiple_subscribers(self):
-        with tornado_server():
+        @csp.node
+        def send_on_status(status: ts[Status], uri: str, val: str) -> ts[str]:
+            if csp.ticked(status):
+                if uri in status.msg and status.status_code == WebsocketStatus.ACTIVE.value:
+                    return val
 
-            @csp.graph
-            def g():
-                ws = WebsocketAdapterManager(dynamic=True)
-                conn_request1 = csp.const(
-                    ConnectionRequest(uri="ws://localhost:8000/", on_connect_payload="hey world from 8000")
-                )
-                conn_request2 = csp.const(
-                    ConnectionRequest(uri="ws://localhost:8001/", on_connect_payload="hey world from 8001")
-                )
-                recv = ws.subscribe(str, RawTextMessageMapper(), connection_request=conn_request1)
-                recv2 = ws.subscribe(str, RawTextMessageMapper(), connection_request=conn_request2)
-
-                csp.add_graph_output("recv", recv)
-                csp.add_graph_output("recv2", recv2)
-
-                merged = csp.flatten([recv, recv2])
-                stop = csp.filter(csp.count(merged) == 2, merged)
-                csp.stop_engine(stop)
-
-            msgs = csp.run(g, starttime=datetime.now(pytz.UTC), endtime=timedelta(seconds=5), realtime=True)
-            assert len(msgs["recv"]) == 1
-            assert msgs["recv"][0][1].msg == "hey world from 8000"
-            assert msgs["recv"][0][1].uri == "ws://localhost:8000/"
-            assert len(msgs["recv2"]) == 1
-            assert msgs["recv2"][0][1].msg == "hey world from 8001"
-            assert msgs["recv2"][0][1].uri == "ws://localhost:8001/"
+        with tornado_server():
+            # We do this to only spawn the tornado server once for both options
+            for use_on_connect_payload in [True, False]:
+
+                @csp.graph
+                def g():
+                    ws = WebsocketAdapterManager(dynamic=True)
+                    if use_on_connect_payload:
+                        conn_request1 = csp.const(
+                            ConnectionRequest(uri="ws://localhost:8000/", on_connect_payload="hey world from 8000")
+                        )
+                        conn_request2 = csp.const(
+                            ConnectionRequest(uri="ws://localhost:8001/", on_connect_payload="hey world from 8001")
+                        )
+                    else:
+                        conn_request1 = csp.const(ConnectionRequest(uri="ws://localhost:8000/"))
+                        conn_request2 = csp.const(ConnectionRequest(uri="ws://localhost:8001/"))
+                        status = ws.status()
+                        to_send = send_on_status(status, "ws://localhost:8000/", "hey world from 8000")
+                        to_send2 = send_on_status(status, "ws://localhost:8001/", "hey world from 8001")
+                        ws.send(to_send, connection_request=conn_request1)
+                        ws.send(to_send2, connection_request=conn_request2)
+
+                    recv = ws.subscribe(str, RawTextMessageMapper(), connection_request=conn_request1)
+                    recv2 = ws.subscribe(str, RawTextMessageMapper(), connection_request=conn_request2)
+
+                    csp.add_graph_output("recv", recv)
+                    csp.add_graph_output("recv2", recv2)
+
+                    merged = csp.flatten([recv, recv2])
+                    stop = csp.filter(csp.count(merged) == 2, merged)
+                    csp.stop_engine(stop)
+
+                msgs = csp.run(g, starttime=datetime.now(pytz.UTC), endtime=timedelta(seconds=5), realtime=True)
+                assert len(msgs["recv"]) == 1
+                assert msgs["recv"][0][1].msg == "hey world from 8000"
+                assert msgs["recv"][0][1].uri == "ws://localhost:8000/"
+                assert len(msgs["recv2"]) == 1
+                assert msgs["recv2"][0][1].msg == "hey world from 8001"
+                assert msgs["recv2"][0][1].uri == "ws://localhost:8001/"
 
     @pytest.mark.parametrize("dynamic", [False, True])
     def test_send_recv_burst_json(self, dynamic):