Set default prefetch_count when create channel for AMQP queue listener

tests/test_configs.py

@@ -1,6 +1,9 @@
 from decimal import Decimal
 
+import pytest
+
 from tomodachi.config import merge_dicts, parse_config_files
+from tomodachi.helpers.dict import get_item_by_path
 
 
 def test_merge_dicts() -> None:
@@ -34,3 +37,25 @@ def test_parse_config_file() -> None:
 def test_parse_no_config_file() -> None:
     result = parse_config_files([])
     assert result is None
+
+
+def test_get_item_by_path() -> None:
+    context = {
+        "options": {
+            "http": {"port": 4711, "access_log": True},
+            "amqp": {
+                "login": "guest",
+                "password": "guest",
+                "qos": {
+                    "queue_prefetch_count": 150,
+                },
+            },
+        }
+    }
+    assert get_item_by_path(context, "options.http.port") == 4711
+    assert get_item_by_path(context, "options.amqp.qos.queue_prefetch_count", 100) == 150
+    assert get_item_by_path(context, "options.amqp.qos.global_prefetch_count", 400) == 400
+    with pytest.raises(KeyError):
+        get_item_by_path(context, "options http")
+    with pytest.raises(ValueError):
+        get_item_by_path(context, "options.http.port.access_log")

tomodachi/helpers/dict.py

@@ -1,4 +1,4 @@
-from typing import Dict
+from typing import Any, Dict
 
 
 def merge_dicts(dict1: Dict, dict2: Dict) -> Dict:
@@ -22,3 +22,20 @@ def merge_dicts(dict1: Dict, dict2: Dict) -> Dict:
             context[k] = v2
 
     return context
+
+
+def get_item_by_path(dict: Dict, path: str, default: Any = None) -> Any:
+    if "." not in path:
+        raise KeyError("Key path must contain '.' ")
+    doted_paths = path.split(".")
+    item = dict.get(doted_paths[0], {})
+    for i in range(1, len(doted_paths)):
+        if i == len(doted_paths) - 1:
+            default_value = default
+        else:
+            default_value = {}
+        if isinstance(item, Dict):
+            item = item.get(doted_paths[i], default_value)
+        else:
+            raise ValueError("Item at key path {} is not a Dict".format(".".join(doted_paths[:i])))
+    return item

tomodachi/transport/amqp.py

@@ -10,7 +10,7 @@
 
 import aioamqp
 
-from tomodachi.helpers.dict import merge_dicts
+from tomodachi.helpers.dict import get_item_by_path, merge_dicts
 from tomodachi.helpers.execution_context import (
     decrease_execution_context_value,
     increase_execution_context_value,
@@ -432,6 +432,10 @@ async def subscribe(cls: Any, obj: Any, context: Dict) -> Optional[Callable]:
 
         cls.channel = None
         channel = await cls.connect(cls, obj, context)
+        queue_prefetch_count = get_item_by_path(context, "options.amqp.qos.queue_prefetch_count", 100)
+        global_prefetch_count = get_item_by_path(context, "options.amqp.qos.global_prefetch_count", 400)
+        await channel.basic_qos(prefetch_count=queue_prefetch_count, prefetch_size=0, connection_global=False)
+        await channel.basic_qos(prefetch_count=global_prefetch_count, prefetch_size=0, connection_global=True)
 
         async def _subscribe() -> None:
             async def declare_queue(