fix: async streaming handler send to queue logic

dynamiq/callbacks/streaming.py

@@ -61,7 +61,7 @@ def on_node_execute_stream(
             data=format_value(chunk),
             event=serialized.get("streaming", {}).get("event"),
         )
-        self.queue.put_nowait(event)
+        self.send_to_queue(event)
 
     def on_workflow_end(
         self, serialized: dict[str, Any], output_data: dict[str, Any], **kwargs: Any
@@ -80,7 +80,7 @@ def on_workflow_end(
             data=format_value(output_data),
             event=serialized.get("streaming", {}).get("event"),
         )
-        self.queue.put_nowait(event)
+        self.send_to_queue(event)
         self.done_event.set()
 
     def on_workflow_error(
@@ -95,6 +95,10 @@ def on_workflow_error(
         """
         self.done_event.set()
 
+    def send_to_queue(self, event: StreamingEventMessage):
+        """Send the event to the queue."""
+        self.queue.put_nowait(event)
+
 
 class StreamingIteratorCallbackHandler(StreamingQueueCallbackHandler):
     """Callback handler for streaming events using an iterator."""
@@ -154,19 +158,26 @@ def __init__(
         self,
         queue: asyncio.Queue | None = None,
         done_event: asyncio.Event | None = None,
+        loop: asyncio.AbstractEventLoop | None = None,
     ) -> None:
         """Initialize AsyncStreamingIteratorCallbackHandler.
 
         Args:
             queue (asyncio.Queue | None): Queue for streaming events.
             done_event (asyncio.Event | None): Event to signal completion.
+            loop (asyncio.AbstractEventLoop | None): Event loop.
         """
         if queue is None:
             queue = asyncio.Queue()
         if done_event is None:
             done_event = asyncio.Event()
         super().__init__(queue, done_event)
         self._iterator = self._iter_queue_events()
+        self.loop = loop or asyncio.get_event_loop()
+
+    def send_to_queue(self, event: StreamingEventMessage):
+        """Send the event to the queue."""
+        asyncio.run_coroutine_threadsafe(self.queue.put(event), self.loop)
 
     async def _iter_queue_events(self) -> AsyncIterator[StreamingEventMessage]:
         """Async iterate over queue events.