added lsiten to port 7447 if its not already allocated for in contianer usecases

Author: PhillipTh0mas

make87/interfaces/zenoh/interface.py

@@ -2,6 +2,7 @@
 import logging
 from typing import Any, Callable, Optional, Union
 import zenoh
+import socket
 from functools import cached_property
 from make87.interfaces.base import InterfaceBase
 from make87.interfaces.zenoh.model import (
@@ -11,6 +12,7 @@
     ZenohProviderConfig,
 )
 
+logger = logging.getLogger(__name__)
 
 class ZenohInterface(InterfaceBase):
     """
@@ -21,6 +23,10 @@ class ZenohInterface(InterfaceBase):
     @cached_property
     def zenoh_config(self) -> zenoh.Config:
         cfg = zenoh.Config()
+
+        if not is_port_in_use(7447):
+            cfg.insert_json5("listen/endpoints", json.dumps(["tcp/0.0.0.0:7447"]))
+
         endpoints = {
             f"tcp/{x.vpn_ip}:{x.vpn_port}"
             for x in list(self.interface_config.requesters.values()) + list(self.interface_config.subscribers.values())
@@ -125,3 +131,14 @@ def get_server(self, name: str) -> Any:
         Zenoh does not have a server concept, so this method is not implemented.
         """
         raise NotImplementedError("Zenoh does not support server interfaces.")
+
+
+def is_port_in_use(port: int, host: str = "0.0.0.0") -> bool:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        try:
+            sock.bind((host, port))
+            return False  # Not in use
+        except OSError:
+            logger.info(f"Port {port} is already in use on {host}.")
+            return True   # Already bound

tests/integration/zenoh/test_pub_sub.py

@@ -56,7 +56,7 @@ def test_pub_sub_combination(priority, reliability, congestion_control, express,
                             handler_type=handler_type,
                             capacity=handler_capacity,
                         ),
-                        vpn_ip="172.0.0.1",
+                        vpn_ip="localhost",
                         vpn_port=7447,
                         same_node=True,
                     ),
@@ -129,19 +129,19 @@ def test_pub_sub_combination(priority, reliability, congestion_control, express,
         }
     )
 
-    # Start subscriber
-    subscriber_proc = subprocess.Popen(
-        [sys.executable, str(subscriber_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=subscriber_env
-    )
-
-    time.sleep(1)  # Let subscriber boot up
 
     # Start publisher (non-blocking)
     publisher_proc = subprocess.Popen(
         [sys.executable, str(publisher_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=publisher_env
     )
 
     time.sleep(1)  # Let publisher publish + subscriber receive
+    # Start subscriber
+    subscriber_proc = subprocess.Popen(
+        [sys.executable, str(subscriber_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=subscriber_env
+    )
+
+    time.sleep(1)  # Let subscriber boot up
 
     # Kill publisher
     publisher_proc.terminate()
@@ -189,7 +189,7 @@ def test_defaults_only():
                         topic_key="my_topic_key",
                         protocol="zenoh",
                         message_type="make87_messages.text.text_plain.PlainText",
-                        vpn_ip="127.0.0.1",
+                        vpn_ip="localhost",
                         vpn_port=7447,
                         same_node=True,
                     ),
@@ -258,24 +258,24 @@ def test_defaults_only():
         }
     )
 
-    # Start subscriber
-    subscriber_proc = subprocess.Popen(
-        [sys.executable, str(subscriber_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=subscriber_env
-    )
-
-    time.sleep(1)  # Let subscriber boot up
 
     # Start publisher (non-blocking)
     publisher_proc = subprocess.Popen(
         [sys.executable, str(publisher_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=publisher_env
     )
 
     time.sleep(1)  # Let publisher publish + subscriber receive
+    # Start subscriber
+    subscriber_proc = subprocess.Popen(
+        [sys.executable, str(subscriber_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=subscriber_env
+    )
+
+    time.sleep(1)  # Let subscriber boot up
 
     # Kill publisher
     publisher_proc.terminate()
     try:
-        publisher_proc.communicate(timeout=5)
+        pub_stdout, pub_stderr = publisher_proc.communicate(timeout=5)
     except subprocess.TimeoutExpired:
         publisher_proc.kill()
         publisher_proc.communicate()