ref(worker): Add queue as optional to allow for initialisation in start

GH-4581

Author: srothh

sentry_sdk/worker.py

@@ -191,7 +191,7 @@ def _target(self) -> None:
 
 class AsyncWorker(Worker):
     def __init__(self, queue_size: int = DEFAULT_QUEUE_SIZE) -> None:
-        self._queue: asyncio.Queue[Any] = asyncio.Queue(maxsize=queue_size)
+        self._queue: Optional[asyncio.Queue[Any]] = None
         self._queue_size = queue_size
         self._task: Optional[asyncio.Task[None]] = None
         # Event loop needs to remain in the same process
@@ -210,10 +210,11 @@ def is_alive(self) -> bool:
 
     def kill(self) -> None:
         if self._task:
-            try:
-                self._queue.put_nowait(_TERMINATOR)
-            except asyncio.QueueFull:
-                logger.debug("async worker queue full, kill failed")
+            if self._queue is not None:
+                try:
+                    self._queue.put_nowait(_TERMINATOR)
+                except asyncio.QueueFull:
+                    logger.debug("async worker queue full, kill failed")
             # Also cancel any active callback tasks
             # Avoid modifying the set while cancelling tasks
             tasks_to_cancel = set(self._active_tasks)
@@ -240,14 +241,16 @@ def start(self) -> None:
                 self._task_for_pid = None
 
     def full(self) -> bool:
+        if self._queue is None:
+            return True
         return self._queue.full()
 
     def _ensure_task(self) -> None:
         if not self.is_alive:
             self.start()
 
     async def _wait_flush(self, timeout: float, callback: Optional[Any] = None) -> None:
-        if not self._loop or not self._loop.is_running():
+        if not self._loop or not self._loop.is_running() or self._queue is None:
             return
 
         initial_timeout = min(0.1, timeout)
@@ -276,14 +279,17 @@ async def flush_async(self, timeout: float, callback: Optional[Any] = None) -> N
 
     def submit(self, callback: Callable[[], Any]) -> bool:
         self._ensure_task()
-
+        if self._queue is None:
+            return False
         try:
             self._queue.put_nowait(callback)
             return True
         except asyncio.QueueFull:
             return False
 
     async def _target(self) -> None:
+        if self._queue is None:
+            return
         while True:
             callback = await self._queue.get()
             if callback is _TERMINATOR:
@@ -310,5 +316,6 @@ def _on_task_complete(self, task: asyncio.Task[None]) -> None:
         finally:
             # Mark the task as done and remove it from the active tasks set
             # This happens only after the task has completed
-            self._queue.task_done()
+            if self._queue is not None:
+                self._queue.task_done()
             self._active_tasks.discard(task)