feat: Rewrite parameter and cli arguments handling in Rosbridge server

rosbridge_library/src/rosbridge_library/capabilities/advertise.py

@@ -30,6 +30,7 @@
 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 # POSSIBILITY OF SUCH DAMAGE.
+from __future__ import annotations
 
 import fnmatch
 
@@ -83,7 +84,7 @@ class Advertise(Capability):
     advertise_msg_fields = ((True, "topic", str), (True, "type", str))
     unadvertise_msg_fields = ((True, "topic", str),)
 
-    topics_glob = None
+    topics_glob: list[str] | None = None
 
     def __init__(self, protocol):
         # Call superclass constructor
@@ -96,8 +97,11 @@ def __init__(self, protocol):
         # Initialize class variables
         self._registrations = {}
 
-        if protocol.parameters and "unregister_timeout" in protocol.parameters:
-            manager.unregister_timeout = protocol.parameters.get("unregister_timeout")
+        if protocol.parameters:
+            if "unregister_timeout" in protocol.parameters:
+                manager.unregister_timeout = protocol.parameters["unregister_timeout"]
+            if "topics_glob" in protocol.parameters:
+                self.topics_glob = protocol.parameters["topics_glob"]
 
     def advertise(self, message):
         # Pull out the ID
@@ -109,10 +113,10 @@ def advertise(self, message):
         latch = message.get("latch", False)
         queue_size = message.get("queue_size", 100)
 
-        if Advertise.topics_glob is not None and Advertise.topics_glob:
+        if self.topics_glob:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
-            for glob in Advertise.topics_glob:
+            for glob in self.topics_glob:
                 if fnmatch.fnmatch(topic, glob):
                     self.protocol.log(
                         "debug",
@@ -144,10 +148,10 @@ def unadvertise(self, message):
         self.basic_type_check(message, self.unadvertise_msg_fields)
         topic = message["topic"]
 
-        if Advertise.topics_glob is not None and Advertise.topics_glob:
+        if self.topics_glob:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
-            for glob in Advertise.topics_glob:
+            for glob in self.topics_glob:
                 if fnmatch.fnmatch(topic, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/advertise_action.py

@@ -29,9 +29,10 @@
 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 # POSSIBILITY OF SUCH DAMAGE.
+from __future__ import annotations
 
 import fnmatch
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from action_msgs.msg import GoalStatus
 from rclpy.action import ActionServer
@@ -42,7 +43,9 @@
 from rosbridge_library.capability import Capability
 from rosbridge_library.internal import message_conversion
 from rosbridge_library.internal.ros_loader import get_action_class
-from rosbridge_library.protocol import Protocol
+
+if TYPE_CHECKING:
+    from rosbridge_library.protocol import Protocol
 
 
 class AdvertisedActionHandler:
@@ -196,14 +199,17 @@ def graceful_shutdown(self) -> None:
 
 
 class AdvertiseAction(Capability):
-    actions_glob = None
-
     advertise_action_msg_fields = ((True, "action", str), (True, "type", str))
 
+    actions_glob: list[str] | None = None
+
     def __init__(self, protocol: Protocol) -> None:
         # Call superclass constructor
         Capability.__init__(self, protocol)
 
+        if protocol.parameters and "actions_glob" in protocol.parameters:
+            self.actions_glob = protocol.parameters["actions_glob"]
+
         # Register the operations that this capability provides
         protocol.register_operation("advertise_action", self.advertise_action)
 
@@ -214,13 +220,13 @@ def advertise_action(self, message: dict) -> None:
         # parse the incoming message
         action_name = message["action"]
 
-        if AdvertiseAction.actions_glob is not None and AdvertiseAction.actions_glob:
+        if self.actions_glob:
             self.protocol.log(
                 "debug",
                 "Action security glob enabled, checking action: " + action_name,
             )
             match = False
-            for glob in AdvertiseAction.actions_glob:
+            for glob in self.actions_glob:
                 if fnmatch.fnmatch(action_name, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/advertise_service.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import fnmatch
 
 import rclpy
@@ -85,14 +87,17 @@ def graceful_shutdown(self):
 
 
 class AdvertiseService(Capability):
-    services_glob = None
-
     advertise_service_msg_fields = ((True, "service", str), (True, "type", str))
 
+    services_glob: list[str] | None = None
+
     def __init__(self, protocol):
         # Call superclass constructor
         Capability.__init__(self, protocol)
 
+        if protocol.parameters and "services_glob" in protocol.parameters:
+            self.services_glob = protocol.parameters["services_glob"]
+
         # Register the operations that this capability provides
         protocol.register_operation("advertise_service", self.advertise_service)
 
@@ -103,13 +108,13 @@ def advertise_service(self, message):
         # parse the incoming message
         service_name = message["service"]
 
-        if AdvertiseService.services_glob is not None and AdvertiseService.services_glob:
+        if self.services_glob:
             self.protocol.log(
                 "debug",
                 "Service security glob enabled, checking service: " + service_name,
             )
             match = False
-            for glob in AdvertiseService.services_glob:
+            for glob in self.services_glob:
                 if fnmatch.fnmatch(service_name, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/call_service.py

@@ -29,6 +29,7 @@
 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 # POSSIBILITY OF SUCH DAMAGE.
+from __future__ import annotations
 
 import fnmatch
 from functools import partial
@@ -45,25 +46,27 @@ class CallService(Capability):
         (False, "compression", str),
     )
 
-    services_glob = None
+    # parameters
+    services_glob: list[str] | None = None
+    default_timeout: float = 5.0
+    call_services_in_new_thread: bool = True
 
     def __init__(self, protocol):
         # Call superclass constructor
         Capability.__init__(self, protocol)
 
-        self.default_timeout = (
-            protocol.node_handle.get_parameter("default_call_service_timeout")
-            .get_parameter_value()
-            .double_value
-        )
+        if self.protocol.parameters:
+            if "services_glob" in self.protocol.parameters:
+                self.services_glob = self.protocol.parameters["services_glob"]
+            if "default_call_service_timeout" in self.protocol.parameters:
+                self.default_timeout = self.protocol.parameters["default_call_service_timeout"]
+            if "call_services_in_new_thread" in self.protocol.parameters:
+                self.call_services_in_new_thread = self.protocol.parameters[
+                    "call_services_in_new_thread"
+                ]
 
         # Register the operations that this capability provides
-        call_services_in_new_thread = (
-            protocol.node_handle.get_parameter("call_services_in_new_thread")
-            .get_parameter_value()
-            .bool_value
-        )
-        if call_services_in_new_thread:
+        if self.call_services_in_new_thread:
             # Calls the service in a separate thread so multiple services can be processed simultaneously.
             protocol.node_handle.get_logger().info("Calling services in new thread")
             protocol.register_operation(
@@ -88,12 +91,12 @@ def call_service(self, message):
         args = message.get("args", [])
         timeout = message.get("timeout", self.default_timeout)
 
-        if CallService.services_glob is not None and CallService.services_glob:
+        if self.services_glob:
             self.protocol.log(
                 "debug", "Service security glob enabled, checking service: " + service
             )
             match = False
-            for glob in CallService.services_glob:
+            for glob in self.services_glob:
                 if fnmatch.fnmatch(service, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/publish.py

@@ -30,6 +30,7 @@
 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 # POSSIBILITY OF SUCH DAMAGE.
+from __future__ import annotations
 
 import fnmatch
 
@@ -40,7 +41,7 @@
 class Publish(Capability):
     publish_msg_fields = ((True, "topic", str),)
 
-    topics_glob = None
+    topics_glob: list[str] | None = None
 
     def __init__(self, protocol):
         # Call superclass constructor
@@ -52,8 +53,11 @@ def __init__(self, protocol):
         # Save the topics that are published on for the purposes of unregistering
         self._published = {}
 
-        if protocol.parameters and "unregister_timeout" in protocol.parameters:
-            manager.unregister_timeout = protocol.parameters.get("unregister_timeout")
+        if protocol.parameters:
+            if "unregister_timeout" in protocol.parameters:
+                manager.unregister_timeout = protocol.parameters["unregister_timeout"]
+            if "topics_glob" in protocol.parameters:
+                self.topics_glob = protocol.parameters["topics_glob"]
 
     def publish(self, message):
         # Do basic type checking
@@ -62,10 +66,10 @@ def publish(self, message):
         latch = message.get("latch", False)
         queue_size = message.get("queue_size", 100)
 
-        if Publish.topics_glob is not None and Publish.topics_glob:
+        if self.topics_glob:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
-            for glob in Publish.topics_glob:
+            for glob in self.topics_glob:
                 if fnmatch.fnmatch(topic, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/send_action_goal.py

@@ -56,7 +56,9 @@ class SendActionGoal(Capability):
     )
     cancel_action_goal_msg_fields = ((True, "action", str),)
 
-    actions_glob = None
+    actions_glob: list[str] | None = None
+    send_action_goals_in_new_thread: bool = False
+
     client_handler_list: dict[str, ActionClientHandler]
 
     def __init__(self, protocol: Protocol) -> None:
@@ -65,13 +67,16 @@ def __init__(self, protocol: Protocol) -> None:
 
         self.client_handler_list = {}
 
+        if protocol.parameters:
+            if "actions_glob" in protocol.parameters:
+                self.actions_glob = protocol.parameters["actions_glob"]
+            if "send_action_goals_in_new_thread" in protocol.parameters:
+                self.send_action_goals_in_new_thread = protocol.parameters[
+                    "send_action_goals_in_new_thread"
+                ]
+
         # Register the operations that this capability provides
-        send_action_goals_in_new_thread = (
-            protocol.node_handle.get_parameter("send_action_goals_in_new_thread")
-            .get_parameter_value()
-            .bool_value
-        )
-        if send_action_goals_in_new_thread:
+        if self.send_action_goals_in_new_thread:
             # Sends the action goal in a separate thread so multiple actions can be processed simultaneously.
             protocol.node_handle.get_logger().info("Sending action goals in new thread")
             protocol.register_operation(
@@ -103,10 +108,10 @@ def send_action_goal(self, message: dict) -> None:
         compression = message.get("compression", "none")
         args = message.get("args", [])
 
-        if SendActionGoal.actions_glob is not None and SendActionGoal.actions_glob:
+        if self.actions_glob:
             self.protocol.log("debug", f"Action security glob enabled, checking action: {action}")
             match = False
-            for glob in SendActionGoal.actions_glob:
+            for glob in self.actions_glob:
                 if fnmatch.fnmatch(action, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/subscribe.py

@@ -29,6 +29,7 @@
 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 # POSSIBILITY OF SUCH DAMAGE.
+from __future__ import annotations
 
 import fnmatch
 from functools import partial
@@ -224,12 +225,15 @@ class Subscribe(Capability):
     )
     unsubscribe_msg_fields = (True, "topic", str)
 
-    topics_glob = None
+    topics_glob: list[str] | None = None
 
     def __init__(self, protocol):
         # Call superclass constructor
         Capability.__init__(self, protocol)
 
+        if protocol.parameters and "topics_glob" in protocol.parameters:
+            self.topics_glob = protocol.parameters["topics_glob"]
+
         # Register the operations that this capability provides
         protocol.register_operation("subscribe", self.subscribe)
         protocol.register_operation("unsubscribe", self.unsubscribe)
@@ -246,10 +250,10 @@ def subscribe(self, msg):
         # Make the subscription
         topic = msg["topic"]
 
-        if Subscribe.topics_glob is not None and Subscribe.topics_glob:
+        if self.topics_glob is not None and self.topics_glob:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
-            for glob in Subscribe.topics_glob:
+            for glob in self.topics_glob:
                 if fnmatch.fnmatch(topic, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/unadvertise_action.py

@@ -29,34 +29,41 @@
 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 # POSSIBILITY OF SUCH DAMAGE.
+from __future__ import annotations
 
 import fnmatch
+from typing import TYPE_CHECKING
 
 from rosbridge_library.capability import Capability
-from rosbridge_library.protocol import Protocol
+
+if TYPE_CHECKING:
+    from rosbridge_library.protocol import Protocol
 
 
 class UnadvertiseAction(Capability):
-    actions_glob = None
+    actions_glob: list[str] | None = None
 
     def __init__(self, protocol: Protocol) -> None:
         # Call superclass constructor
         Capability.__init__(self, protocol)
 
+        if protocol.parameters and "actions_glob" in protocol.parameters:
+            self.actions_glob = protocol.parameters["actions_glob"]
+
         # Register the operations that this capability provides
         protocol.register_operation("unadvertise_action", self.unadvertise_action)
 
     def unadvertise_action(self, message: dict) -> None:
         # parse the message
         action_name = message["action"]
 
-        if UnadvertiseAction.actions_glob is not None and UnadvertiseAction.actions_glob:
+        if self.actions_glob:
             self.protocol.log(
                 "debug",
                 f"Action security glob enabled, checking action: {action_name}",
             )
             match = False
-            for glob in UnadvertiseAction.actions_glob:
+            for glob in self.actions_glob:
                 if fnmatch.fnmatch(action_name, glob):
                     self.protocol.log(
                         "debug",