Merge branch 'dev' into refactor/logging

src/control_backend/main.py

@@ -1,9 +1,11 @@
 import contextlib
 import logging
+import threading
 
 import zmq
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from zmq.asyncio import Context
 
 from control_backend.agents import (
     BeliefCollectorAgent,
@@ -14,12 +16,30 @@ from control_backend.agents import (
 from control_backend.agents.bdi import BDICoreAgent, TBeliefExtractorAgent
 from control_backend.api.v1.router import api_router
 from control_backend.core.config import settings
-from control_backend.core.zmq_context import context
 from control_backend.logging import setup_logging
 
 logger = logging.getLogger(__name__)
 
 
+def setup_sockets():
+    context = Context.instance()
+
+    internal_pub_socket = context.socket(zmq.XPUB)
+    internal_pub_socket.bind(settings.zmq_settings.internal_sub_address)
+    logger.debug("Internal publishing socket bound to %s", internal_pub_socket)
+
+    internal_sub_socket = context.socket(zmq.XSUB)
+    internal_sub_socket.bind(settings.zmq_settings.internal_pub_address)
+    logger.debug("Internal subscribing socket bound to %s", internal_sub_socket)
+    try:
+        zmq.proxy(internal_sub_socket, internal_pub_socket)
+    except zmq.ZMQError:
+        logger.warning("Error while handling PUB/SUB proxy. Closing sockets.")
+    finally:
+        internal_pub_socket.close()
+        internal_sub_socket.close()
+
+
 @contextlib.asynccontextmanager
 async def lifespan(app: FastAPI):
     """
@@ -29,18 +49,16 @@ async def lifespan(app: FastAPI):
     setup_logging()
     logger.info("%s is starting up.", app.title)
 
-    # --- Initialize Sockets ---
-    logger.info("Initializing ZeroMQ sockets.")
-    try:
-        internal_comm_socket = context.socket(zmq.PUB)
-        internal_comm_address = settings.zmq_settings.internal_comm_address
-        logger.debug("Binding internal PUB socket to address: %s", internal_comm_address)
-        internal_comm_socket.bind(internal_comm_address)
-        app.state.internal_comm_socket = internal_comm_socket
-        logger.info("Internal communication socket bound successfully.")
-    except Exception as e:
-        logger.error("Failed to bind internal communication socket: %s", e, exc_info=True)
-        raise
+    # Initiate sockets
+    proxy_thread = threading.Thread(target=setup_sockets)
+    proxy_thread.daemon = True
+    proxy_thread.start()
+
+    context = Context.instance()
+
+    endpoints_pub_socket = context.socket(zmq.PUB)
+    endpoints_pub_socket.connect(settings.zmq_settings.internal_pub_address)
+    app.state.endpoints_pub_socket = endpoints_pub_socket
 
     # --- Initialize Agents ---
     logger.info("Initializing and starting agents.")