Rename max_queue_size to max_queue_threshold in CoroQueue and related components

Author: phettberg

broqer/coro_queue.py

@@ -24,7 +24,7 @@ async def _coro(value):
 
 
 class MaxQueueException(Exception):
-    """ Exception raised when max_queue_size is exceeded in CoroQueue """
+    """ Exception raised when max_queue_threshold is exceeded in CoroQueue """
 
 
 class AsyncMode(Enum):
@@ -42,19 +42,21 @@ class CoroQueue:  # pylint: disable=too-few-public-methods
     """ Schedules the running of a coroutine given on a mode
     :param coro: Coroutine to be scheduled
     :param mode: scheduling mode (see AsyncMode)
-    :param max_queue_size: queue len error threshold, used with AsyncMode.QUEUE
+    :param max_queue_threshold: queue len error threshold,
+                                used with AsyncMode.QUEUE
     """
     def __init__(self, coro, mode=AsyncMode.CONCURRENT,
-                 max_queue_size: int | None = None):
+                 max_queue_threshold: int | None = None):
 
-        if max_queue_size is not None and mode != AsyncMode.QUEUE:
+        if max_queue_threshold is not None and mode != AsyncMode.QUEUE:
             raise ValueError(
-                "max_queue_size can only be used with mode=AsyncMode.QUEUE"
+                "max_queue_threshold can only be used with "
+                "mode=AsyncMode.QUEUE"
             )
 
         self._coro = coro
         self._mode = mode
-        self._max_queue_size = max_queue_size
+        self._max_queue_threshold = max_queue_threshold
 
         # ._last_args is used for LAST_DISTINCT and keeps the last arguments
         self._last_args = None  # type: Optional[Tuple]
@@ -124,11 +126,11 @@ def _start_task(self, args: Tuple, future: asyncio.Future):
     def _handle_done(self, result_future: asyncio.Future, task: asyncio.Task):
         try:
             result = task.result()
-            if self._queue and self._max_queue_size is not None and \
-                    len(self._queue) > self._max_queue_size:
+            if self._queue and self._max_queue_threshold is not None and \
+                    len(self._queue) > self._max_queue_threshold:
                 raise MaxQueueException(
                     "CoroQueue size is %d and exceeds %d" %
-                    (len(self._queue), self._max_queue_size)
+                    (len(self._queue), self._max_queue_threshold)
                 )
         except asyncio.CancelledError:  # happend in INTERRUPT mode
             result_future.set_result(NONE)

broqer/op/map_async.py

@@ -116,7 +116,8 @@ class MapAsync(Operator):  # pylint: disable=too-many-instance-attributes
     :param mode: behavior when a value is currently processed
     :param error_callback: error callback to be registered
     :param unpack: value from emits will be unpacked as (\\*value)
-    :param max_queue_size: queue len error threshold, used with AsyncMode.QUEUE
+    :param max_queue_threshold: queue len error threshold,
+                                used with AsyncMode.QUEUE
     :param \\*\\*kwargs: keyword arguments to be used for calling coro
 
     :ivar scheduled: Publisher emitting the value when coroutine is actually
@@ -125,11 +126,11 @@ class MapAsync(Operator):  # pylint: disable=too-many-instance-attributes
     def __init__(self,
                  coro, *args, mode=AsyncMode.CONCURRENT,
                  error_callback=default_error_handler, unpack: bool = False,
-                 max_queue_size: int | None = None, **kwargs) -> None:
+                 max_queue_threshold: int | None = None, **kwargs) -> None:
         Operator.__init__(self)
         _coro = wrap_coro(coro, unpack, *args, **kwargs)
         self._coro_queue = CoroQueue(
-            _coro, mode=mode, max_queue_size=max_queue_size
+            _coro, mode=mode, max_queue_threshold=max_queue_threshold
         )
         self._error_callback = error_callback
 
@@ -154,18 +155,19 @@ def build_map_async(coro=None, *,
                     mode: AsyncMode = AsyncMode.CONCURRENT,
                     error_callback=default_error_handler,
                     unpack: bool = False,
-                    max_queue_size: int | None = None):
+                    max_queue_threshold: int | None = None):
     """ Decorator to wrap a function to return a Map operator.
 
     :param coro: coroutine to be wrapped
     :param mode: behavior when a value is currently processed
     :param error_callback: error callback to be registered
     :param unpack: value from emits will be unpacked (*value)
-    :param max_queue_size: queue len error threshold, used with AsyncMode.QUEUE
+    :param max_queue_threshold: queue len error threshold,
+                                used with AsyncMode.QUEUE
     """
     def _build_map_async(coro):
         return MapAsync(coro, mode=mode, error_callback=error_callback,
-                        unpack=unpack, max_queue_size=max_queue_size)
+                        unpack=unpack, max_queue_threshold=max_queue_threshold)
 
     if coro:
         return _build_map_async(coro)
@@ -177,32 +179,33 @@ def build_map_async_factory(coro=None, *,
                             mode: AsyncMode = AsyncMode.CONCURRENT,
                             error_callback=default_error_handler,
                             unpack: bool = False,
-                            max_queue_size: int | None = None):
+                            max_queue_threshold: int | None = None):
     """ Decorator to wrap a coroutine to return a factory for MapAsync
         operators.
 
     :param coro: coroutine to be wrapped
     :param mode: behavior when a value is currently processed
     :param error_callback: error callback to be registered
     :param unpack: value from emits will be unpacked (*value)
-    :param max_queue_size: queue len error threshold, used with AsyncMode.QUEUE
+    :param max_queue_threshold: queue len error threshold,
+                                used with AsyncMode.QUEUE
     """
     _mode = mode
 
     def _build_map_async(coro):
         @wraps(coro)
         def _wrapper(*args, mode=None, **kwargs) -> MapAsync:
             if ('unpack' in kwargs) or ('error_callback' in kwargs) or \
-                    ('max_queue_size' in kwargs):
+                    ('max_queue_threshold' in kwargs):
                 raise TypeError(
-                    '"unpack", "error_callback" or "max_queue_size" has to '
-                    'be defined by decorator'
+                    '"unpack", "error_callback" or "max_queue_threshold" '
+                    'has to be defined by decorator'
                 )
             if mode is None:
                 mode = _mode
             return MapAsync(coro, *args, mode=mode, unpack=unpack,
                             error_callback=error_callback,
-                            max_queue_size=max_queue_size, **kwargs)
+                            max_queue_threshold=max_queue_threshold, **kwargs)
         return _wrapper
 
     if coro:

broqer/subscribers/sink_async.py

@@ -47,17 +47,18 @@ class SinkAsync(Subscriber):  # pylint: disable=too-few-public-methods
     :param mode: behavior when a value is currently processed
     :param error_callback: error callback to be registered
     :param unpack: value from emits will be unpacked as (\\*value)
-    :param max_queue_size: queue len error threshold,used with AsyncMode.QUEUE
+    :param max_queue_threshold: queue len error threshold,
+                                used with AsyncMode.QUEUE
     :param \\*\\*kwargs: keyword arguments to be used for calling coro
     """
     def __init__(self, coro, *args, mode=AsyncMode.CONCURRENT,
                  error_callback=default_error_handler,
                  unpack: bool = False,
-                 max_queue_size: int | None = None, **kwargs) -> None:
+                 max_queue_threshold: int | None = None, **kwargs) -> None:
 
         _coro = wrap_coro(coro, unpack, *args, **kwargs)
         self._coro_queue = CoroQueue(
-            _coro, mode=mode, max_queue_size=max_queue_size
+            _coro, mode=mode, max_queue_threshold=max_queue_threshold
         )
         self._error_callback = error_callback
 
@@ -93,29 +94,30 @@ def build_sink_async_factory(coro=None, *,
                              mode: AsyncMode = AsyncMode.CONCURRENT,
                              error_callback=default_error_handler,
                              unpack: bool = False,
-                             max_queue_size: int | None = None):
+                             max_queue_threshold: int | None = None):
     """ Decorator to wrap a coroutine to return a factory for SinkAsync
         subscribers.
 
     :param coro: coroutine to be wrapped
     :param mode: behavior when a value is currently processed
     :param error_callback: error callback to be registered
     :param unpack: value from emits will be unpacked (*value)
-    :param max_queue_size: queue len error threshold, used with AsyncMode.QUEUE
+    :param max_queue_threshold: queue len error threshold,
+                                used with AsyncMode.QUEUE
     """
     def _build_sink_async(coro):
         @wraps(coro)
         def _wrapper(*args, **kwargs) -> SinkAsync:
-            if ('unpack' in kwargs) or ('max_queue_size' in kwargs) or \
+            if ('unpack' in kwargs) or ('max_queue_threshold' in kwargs) or \
                     ('error_callback' in kwargs) or ('mode' in kwargs):
                 raise TypeError(
                     '"unpack", "mode", "error_callback" and '
-                    '"max_queue_size" has to be defined by decorator'
+                    '"max_queue_threshold" has to be defined by decorator'
                 )
 
             return SinkAsync(coro, *args, mode=mode,
                              error_callback=error_callback, unpack=unpack,
-                             max_queue_size=max_queue_size,
+                             max_queue_threshold=max_queue_threshold,
                              **kwargs)
         return _wrapper
 
@@ -129,21 +131,22 @@ def sink_async_property(coro=None, *,
                         mode: AsyncMode = AsyncMode.CONCURRENT,
                         error_callback=default_error_handler,
                         unpack: bool = False,
-                        max_queue_size: int | None = None):
+                        max_queue_threshold: int | None = None):
     """ Decorator to build a property returning a SinkAsync subscriber.
 
     :param coro: coroutine to be wrapped
     :param mode: behavior when a value is currently processed
     :param error_callback: error callback to be registered
     :param unpack: value from emits will be unpacked (*value)
-    :param max_queue_size: queue len error threshold, used with AsyncMode.QUEUE
+    :param max_queue_threshold: queue len error threshold,
+                                used with AsyncMode.QUEUE
     """
     def build_sink_async_property(coro):
         @property
         def _build_sink_async(self):
             return SinkAsync(coro, self, mode=mode,
                              error_callback=error_callback, unpack=unpack,
-                             max_queue_size=max_queue_size)
+                             max_queue_threshold=max_queue_threshold)
         return _build_sink_async
 
     if coro:

tests/test_coro_queue.py

@@ -181,7 +181,7 @@ async def _coro(index):
         await asyncio.sleep(0.001)
         return index + 2
 
-    coro_queue = CoroQueue(coro=_coro, mode=AsyncMode.QUEUE, max_queue_size=10)
+    coro_queue = CoroQueue(coro=_coro, mode=AsyncMode.QUEUE, max_queue_threshold=10)
 
     future_0 = coro_queue.schedule(0)
     future_1 = coro_queue.schedule(40)
@@ -209,9 +209,9 @@ async def _coro():
         return 42
 
     with pytest.raises(ValueError) as exc_info:
-        CoroQueue(coro=_coro, mode=mode, max_queue_size=10)
+        CoroQueue(coro=_coro, mode=mode, max_queue_threshold=10)
 
-    assert str(exc_info.value) == 'max_queue_size can only be used with mode=AsyncMode.QUEUE'
+    assert str(exc_info.value) == 'max_queue_threshold can only be used with mode=AsyncMode.QUEUE'
 
 
 @pytest.mark.parametrize('distinct', [True, False])