Repository: nautechsystems/nautilus_trader
File: nautilus_trader/adapters/interactive_brokers/client/client.py (Python)

Code before:

        Continuously retrieves and executes tasks from `msg_handler_task_queue`, which are
        typically partial functions representing message handling operations received from the ibapi wrapper.
        The method ensures each task is awaited, thereby executing it. After task execution, it marks
        the task as done in the queue.

        This method is designed to run indefinitely until externally cancelled, typically as part
        of an application shutdown or when the handling context changes requiring a halt in operations.

        """
        try:
            while True:
                handler_task = await self._msg_handler_task_queue.get()
                await handler_task()
                # try:
                #     await handler_task()
                # except Exception as e:
                #     exc_type = type(e)
                #     exc_traceback = e.__traceback__
                #     stack_trace = traceback.format_exception(exc_type, e, exc_traceback)
                #     stack_trace_str = "".join(stack_trace)
                #     task_name = getattr(handler_task, "__name__", str(handler_task))
                #     self._log.error(
                #         f"Exception in message handler task '{task_name}': {e!r}\n{stack_trace_str}",
                #     )
                #     raise
                self._msg_handler_task_queue.task_done()
        except asyncio.CancelledError:
            log_msg = f"Handler task processing was cancelled. (qsize={self._msg_handler_task_queue.qsize()})."
            (
                self._log.warning(log_msg)
                if not self._internal_msg_queue.empty()
                else self._log.debug(
                    log_msg,
                )
            )
        finally:
            self._log.debug("Handler task processor stopped")

    def submit_to_msg_handler_queue(self, task: Callable[..., Any]) -> None:
        """
        Submit a task to the message handler's queue for processing.

        This method places a callable task into the message handler task queue,
        ensuring it's scheduled for asynchronous execution according to the queue's
        order. The operation is non-blocking and immediately returns after queueing the task.

        Parameters
        ----------
        task : Callable[..., Any]
            The task to be queued. This task should be a callable that matches

Review comment (bug):
i uncommented this and removed the `raise`. This gives us some helpful error logging, and also won't crash the handling loop

Code after:
        Asynchronously processes handler tasks from the message handler task queue.

        Continuously retrieves and executes tasks from `msg_handler_task_queue`, which are
        typically partial functions representing message handling operations received from the ibapi wrapper.
        The method ensures each task is awaited, thereby executing it. After task execution, it marks
        the task as done in the queue.

        This method is designed to run indefinitely until externally cancelled, typically as part
        of an application shutdown or when the handling context changes requiring a halt in operations.

        """
        try:
            while True:
                handler_task = await self._msg_handler_task_queue.get()
                try:
                    await handler_task()
                except Exception as e:
                    exc_type = type(e)
                    exc_traceback = e.__traceback__
                    stack_trace = traceback.format_exception(exc_type, e, exc_traceback)
                    stack_trace_str = "".join(stack_trace)
                    task_name = getattr(handler_task, "__name__", str(handler_task))
                    self._log.error(
                        f"Exception in message handler task '{task_name}': {e!r}\n{stack_trace_str}",
                    )
                self._msg_handler_task_queue.task_done()
        except asyncio.CancelledError:
            log_msg = f"Handler task processing was cancelled. (qsize={self._msg_handler_task_queue.qsize()})."
            (
                self._log.warning(log_msg)
                if not self._internal_msg_queue.empty()
                else self._log.debug(
                    log_msg,
                )
            )
        finally:
            self._log.debug("Handler task processor stopped")

    def submit_to_msg_handler_queue(self, task: Callable[..., Any]) -> None:
        """
        Submit a task to the message handler's queue for processing.

        This method places a callable task into the message handler task queue,
        ensuring it's scheduled for asynchronous execution according to the queue's
        order. The operation is non-blocking and immediately returns after queueing the task.

        Parameters
        ----------
        task : Callable[..., Any]
            The task to be queued. This task should be a callable that matches
            the expected signature for tasks processed by the message handler.