Commit 978d52c6 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Common - Message Broker - NATS: 

- Improved subscriber error handling and logging
parent 907df502

src/common/message_broker/backend/nats/NatsBackendThread.py

+21 −17
Original line number Diff line number Diff line
@@ -52,9 +52,11 @@ class NatsBackendThread(threading.Thread): 
        self, topic_name : str, timeout : float, out_queue : queue.Queue[Message], unsubscribe : threading.Event,

        ready_event : threading.Event

    ) -> None:

        try:

            LOGGER.info('[_run_subscriber] NATS URI: {:s}'.format(str(self._nats_uri)))

            client = await nats.connect(servers=[self._nats_uri])

        LOGGER.info('[_run_subscriber] Connected!')

            server_version = client.connected_server_version

            LOGGER.info('[_run_subscriber] Connected! NATS Server version: {:s}'.format(str(repr(server_version))))

            subscription = await client.subscribe(topic_name)

            LOGGER.info('[_run_subscriber] Subscribed!')

            ready_event.set()
@@ -68,6 +70,8 @@ class NatsBackendThread(threading.Thread): 
                out_queue.put(Message(message.subject, message.data.decode('UTF-8')))

            await subscription.unsubscribe()

            await client.drain()

        except Exception as e:

            LOGGER.exception('[_run_subscriber] Unhandled Exception')



    def subscribe(

        self, topic_name : str, timeout : float, out_queue : queue.Queue[Message], unsubscribe : threading.Event
@@ -79,7 +83,7 @@ class NatsBackendThread(threading.Thread): 
        self._tasks.append(task)

        LOGGER.info('[subscribe] Waiting for subscriber to be ready...')

        is_ready = ready_event.wait(timeout=120)

        LOGGER.info('[subscribe] Subscriber Ready: {:s}'.format(str(is_ready)))

        LOGGER.info('[subscribe] Subscriber is Ready? {:s}'.format(str(is_ready)))



    def run(self) -> None:

        asyncio.set_event_loop(self._event_loop)