Fixes redis transport

commlib/node.py

@@ -344,8 +344,8 @@ def run_forever(self, sleep_rate: float = 0.01) -> None:
         try:
             while self.state != NodeState.EXITED:
                 time.sleep(sleep_rate)
-        except:
-            pass
+        except Exception as e:
+            self.log.error(f"Exception occurred during run_forever: {str(e)}")
         self.stop()
 
     def stop(self):

commlib/rpc.py

@@ -209,13 +209,14 @@ def run(self) -> None:
         if not self._transport.is_connected and \
             self._state not in (EndpointState.CONNECTED,
                                 EndpointState.CONNECTING):
+            self._transport.start()
             self._main_thread = threading.Thread(target=self.run_forever)
             self._main_thread.daemon = True
             self._t_stop_event = threading.Event()
             self._main_thread.start()
             self._state = EndpointState.CONNECTED
         else:
-            self.logger().debug(
+            self.logger().error(
                 f"Transport already connected - cannot run {self.__class__.__name__}")
 
     def stop(self):

commlib/transports/redis.py

@@ -83,7 +83,7 @@ def __init__(
         super().__init__(*args, **kwargs)
         self._serializer = serializer
         self._compression = compression
-        self.connect()
+        self._connected = False
 
     @property
     def is_connected(self) -> bool:
@@ -236,7 +236,7 @@ def run_forever(self):
 
     def _detach_request_handler(self, payload: str):
         data, header = self._unpack_comm_msg(payload)
-        self.log.info(f"RPC Request <{self._rpc_name}>")
+        # self.log.info(f"RPC Request <{self._rpc_name}>")
         self._on_request_handle(data, header)
 
     def _unpack_comm_msg(self, payload: str) -> Tuple:

examples/simple_rpc/simple_rpc_client.py

@@ -35,6 +35,7 @@ class Response(RPCMessage.Response):
     node = Node(
         node_name="myclient",
         connection_params=conn_params,
+        heartbeats=False,
         # heartbeat_uri='nodes.add_two_ints.heartbeat',
         debug=True,
     )

examples/simple_rpc/simple_rpc_service.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 
 import sys
+import time
 
 from commlib.msg import RPCMessage
 from commlib.node import Node
@@ -40,6 +41,7 @@ def add_two_int_handler(msg):
     node = Node(
         node_name="add_two_ints_node",
         connection_params=conn_params,
+        heartbeats=False,
         # heartbeat_uri='nodes.add_two_ints.heartbeat',
         debug=True,
     )
@@ -50,4 +52,4 @@ def add_two_int_handler(msg):
         on_request=add_two_int_handler,
     )
 
-    node.run_forever(sleep_rate=1)
+    node.run_forever()

requirements.txt

@@ -2,7 +2,7 @@ wheel>=0.38.0
 pydantic>=2.0.0
 ujson>=5.7.0
 rich
-redis[hiredis]
+redis[hiredis]==5.0.1
 paho-mqtt<2.0.0
 pika>=1.3.1
 requests>=2.1.0