Fix duplicate subscription created with wrong 'raw' attribute. (Robot…

…WebTools#798)

rosbridge_library/src/rosbridge_library/internal/subscribers.py

@@ -128,6 +128,7 @@ def __init__(self, topic, client_id, callback, node_handle, msg_type=None, raw=F
         self.node_handle = node_handle
         self.topic = topic
         self.qos = qos
+        self.raw = raw
 
         self.subscriber = node_handle.create_subscription(
             msg_class, topic, self.callback, qos, raw=raw
@@ -174,7 +175,7 @@ def subscribe(self, client_id, callback):
             self.new_subscriptions.update({client_id: callback})
             if self.new_subscriber is None:
                 self.new_subscriber = self.node_handle.create_subscription(
-                    self.msg_class, self.topic, self._new_sub_callback, self.qos
+                    self.msg_class, self.topic, self._new_sub_callback, self.qos, raw=self.raw
                 )
 
     def unsubscribe(self, client_id):

rosbridge_server/CMakeLists.txt

@@ -28,4 +28,5 @@ if(BUILD_TESTING)
   add_launch_test(test/websocket/smoke.test.py)
   add_launch_test(test/websocket/transient_local_publisher.test.py)
   add_launch_test(test/websocket/best_effort_publisher.test.py)
+  add_launch_test(test/websocket/multiple_subscribers_raw.test.py)
 endif()

rosbridge_server/test/websocket/common.py

@@ -38,7 +38,7 @@ def sendJson(self, msg_dict, *, times=1):
 
     def onMessage(self, payload, binary):
         print(f"WebSocket client received message: {payload}")
-        self.message_handler(json.loads(payload))
+        self.message_handler(payload if binary else json.loads(payload))
 
 
 def _generate_node():

rosbridge_server/test/websocket/multiple_subscribers_raw.test.py

@@ -0,0 +1,50 @@
+import os
+import sys
+import unittest
+
+from rclpy.node import Node
+from std_msgs.msg import String
+from twisted.python import log
+
+sys.path.append(os.path.dirname(__file__))  # enable importing from common.py in this directory
+
+import common  # noqa: E402
+from common import expect_messages, sleep, websocket_test  # noqa: E402
+
+log.startLogging(sys.stderr)
+
+generate_test_description = common.generate_test_description
+
+
+class TestMultipleSubscribers(unittest.TestCase):
+    @websocket_test
+    async def test_multiple_subscribers(self, node: Node, make_client):
+        sub_operation_json = {
+            "op": "subscribe",
+            "topic": "/a_topic",
+            "type": "std_msgs/String",
+            "compression": "cbor-raw",
+        }
+        ws_client1 = await make_client()
+        ws_client1.sendJson(sub_operation_json)
+        ws_client2 = await make_client()
+        ws_client2.sendJson(sub_operation_json)
+
+        ws1_completed_future, ws_client1.message_handler = expect_messages(
+            1, "WebSocket 1", node.get_logger()
+        )
+        ws1_completed_future.add_done_callback(lambda _: node.executor.wake())
+        ws2_completed_future, ws_client2.message_handler = expect_messages(
+            1, "WebSocket 2", node.get_logger()
+        )
+        ws2_completed_future.add_done_callback(lambda _: node.executor.wake())
+
+        pub_a = node.create_publisher(String, "/a_topic", 1)
+
+        await sleep(node, 1)
+        pub_a.publish(String(data="hello"))
+        await sleep(node, 1)
+
+        self.assertTrue(await ws1_completed_future)
+        self.assertTrue(await ws2_completed_future)
+        node.destroy_publisher(pub_a)