Fix AsyncEmitSource reuse

mlrun · Dec 26, 2023 · ebf1bed · ebf1bed
1 parent 1cb2ad1
commit ebf1bed
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/storey/sources.py b/storey/sources.py
@@ -583,12 +583,11 @@ def __init__(
     ):
         super().__init__(**kwargs)
         if buffer_size is None:
-            buffer_size = 8
+            self._buffer_size = 8
         elif buffer_size <= 0:
             raise ValueError("Buffer size must be positive")
         else:
-            kwargs["buffer_size"] = buffer_size
-        self._q = asyncio.Queue(buffer_size)
+            kwargs["buffer_size"] = self._buffer_size
         self._key_field = key_field
         self._max_events_before_commit = max_events_before_commit or 20000
         self._max_time_before_commit = max_time_before_commit or 45
@@ -601,6 +600,7 @@ def _init(self):
         super()._init()
         self._is_terminated = False
         self._outstanding_offsets = defaultdict(list)
+        self._q = asyncio.Queue(self._buffer_size)
 
     async def _run_loop(self):
         committer = None