Mqtt broker

CONTRIBUTING.md

@@ -130,6 +130,12 @@ services:
     image: softwaremill/elasticmq-native
     ports:
       - 9324:9324
+
+  mqtt:
+    image: eclipse-mosquitto
+    ports:
+      - 1883:1883
+      - 9001:9001
 ```
 
 ```bash

docs/docs/en/contributing/2_contributing-index.md

@@ -131,6 +131,12 @@ services:
     image: softwaremill/elasticmq-native
     ports:
       - 9324:9324
+
+  mqtt:
+    image: eclipse-mosquitto
+    ports:
+      - 1883:1883
+      - 9001:9001
 ```
 
 ```bash

docs/docs/ru/contributing/2_contributing-index.md

@@ -139,6 +139,12 @@ services:
     image: softwaremill/elasticmq-native
     ports:
       - 9324:9324
+
+  mqtt:
+    image: eclipse-mosquitto
+    ports:
+      - 1883:1883
+      - 9001:9001
 ```
 
 #### Hatch

propan/__about__.py

@@ -42,3 +42,7 @@ def import_error(error_msg: str) -> Mock:
 INSTALL_REDIS = import_error(
     ("\nYou should install RabbitMQ dependencies" '\npip install "propan[async-redis]"')
 )
+
+INSTALL_MQTT = import_error(
+    ("\nYou should install MQTT dependencies" '\npip install "propan[async-mqtt]"')
+)

propan/__init__.py

@@ -31,8 +31,14 @@
 except ImportError:
     SQSBroker = SQSRouter = about.INSTALL_SQS  # type: ignore
 
+try:
+    from propan.brokers.mqtt import MqttBroker
+except ImportError as e:
+    print(e)
+    MqttBroker = about.INSTALL_MQTT  # type: ignore
+
 assert any(
-    (RabbitBroker, NatsBroker, RedisBroker, SQSBroker, KafkaBroker)
+    (RabbitBroker, NatsBroker, RedisBroker, SQSBroker, KafkaBroker, MqttBroker)
 ), about.INSTALL_MESSAGE
 
 __all__ = (  # noqa: F405
@@ -68,4 +74,6 @@
     ## sqs
     "SQSBroker",
     "SQSRouter",
+    ## mqtt
+    "MqttBroker",
 )

propan/brokers/_model/broker_usecase.py

@@ -24,7 +24,6 @@
 from fast_depends._compat import PYDANTIC_V2
 from fast_depends.core import CallModel, build_call_model
 from fast_depends.dependencies import Depends
-from fast_depends.utils import args_to_kwargs
 from typing_extensions import Self, TypeAlias, TypeVar
 
 from propan.brokers._model.routing import BrokerRouter
@@ -150,12 +149,16 @@ def __init__(
 
     def _resolve_connection_kwargs(self, *args: Any, **kwargs: AnyDict) -> AnyDict:
         arguments = get_function_positional_arguments(self.__init__)  # type: ignore
-        init_kwargs = args_to_kwargs(
-            arguments,
-            *self._connection_args,
+
+        init_kwargs = {
             **self._connection_kwargs,
-        )
-        connect_kwargs = args_to_kwargs(arguments, *args, **kwargs)
+            **dict(zip(arguments, self._connection_args)),
+        }
+
+        connect_kwargs = {
+            **kwargs,
+            **dict(zip(arguments, args)),
+        }
         return {**init_kwargs, **connect_kwargs}
 
     @staticmethod

propan/brokers/mqtt/__init__.py

@@ -0,0 +1,5 @@
+from propan.brokers.mqtt.mqtt_broker import MqttBroker
+
+__all__ = (
+    "MqttBroker",
+)

propan/brokers/mqtt/mqtt_broker.py

@@ -0,0 +1,107 @@
+from typing import Any, Callable, Optional, TypeVar, Type, List
+from types import TracebackType
+from functools import wraps
+
+from paho.mqtt.client import Client
+from propan.brokers._model import BrokerAsyncUsecase
+from propan.brokers._model.schemas import PropanMessage
+from propan.brokers.mqtt.schemas import Handler
+from propan.types import SendableMessage, AnyDict
+
+T = TypeVar("T")
+
+
+class MqttBroker(BrokerAsyncUsecase[Any, Client]):
+    handlers: List[Handler]
+
+    _connection: Client
+    _polling_interval: float
+
+    def __init__(
+        self,
+        url: str = "localhost",
+        *,
+        polling_interval: float = 1.0,
+        log_fmt: Optional[str] = None,
+        protocol: str = "mqtt",
+        **kwargs: AnyDict,
+    ) -> None:
+        super().__init__(url, log_fmt=log_fmt, url_=url, protocol=protocol, **kwargs)
+        self._polling_interval = polling_interval
+
+    async def _connect(self, url: str, **kwargs: AnyDict) -> Client:
+        client = Client()
+        client.connect(url, **kwargs)
+        return client
+
+    async def close(  # type: ignore[override]
+        self,
+        exc_type: Optional[Type[BaseException]] = None,
+        exc_val: Optional[BaseException] = None,
+        exec_tb: Optional[TracebackType] = None,
+    ) -> None:
+        await super().close(exc_type, exc_val, exec_tb)
+        if self._connection is not None:
+            self._connection.disconnect()
+            self._connection = None
+
+    async def start(self) -> None:
+        await super().start()
+
+        for h in self.handlers:
+            self._connection.subscribe(h.topic)
+            self._connection.message_callback_add(h.topic, h.callback)
+
+        self._connection.loop_start()
+
+    def handle(self, topic: str):
+        def wrapper(
+            func: Callable[[PropanMessage], Any]
+        ) -> Callable[[PropanMessage], Any]:
+            func = wrap_mqtt(func)
+            # func, dependant = self._wrap_handler(func)
+            handler = Handler(
+                callback=func,
+                # dependant=dependant,
+                topic=topic,
+            )
+            self.handlers.append(handler)
+            return func
+
+        return wrapper
+
+    async def _parse_message(self, message) -> PropanMessage:
+        client, userdata, msg = message
+        return PropanMessage(
+            body=msg.body,
+            raw_message=message,
+        )
+
+    async def _process_message(
+        self,
+        func,
+        watcher,
+    ):
+        @wraps(func)
+        async def wrapper(message: PropanMessage[Any]) -> T:
+            r = await func(message)
+            return r
+        return wrapper
+
+    async def publish(
+        self,
+        message: SendableMessage,
+        *args: Any,
+        callback: bool = False,
+        callback_timeout: Optional[float] = None,
+        raise_timeout: bool = False,
+        **kwargs: Any,
+    ) -> Any:
+        self._connection.publish()
+
+
+def wrap_mqtt(func):
+    @wraps(func)
+    def wrap(client, userdata, msg):
+        return (client, userdata, msg)
+    return wrap

propan/brokers/mqtt/schemas.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+
+from fast_depends.core import CallModel
+
+from propan.brokers._model.schemas import BaseHandler
+from propan.types import DecoratedCallable
+
+
+@dataclass
+class Handler(BaseHandler):
+    topic: str
+
+    def __init__(
+        self,
+        callback: DecoratedCallable,
+        dependant: CallModel,
+        topic: str = "",
+        _description: str = "",
+    ):
+        self.callback = callback
+        self.dependant = dependant
+        self._description = _description
+        self.topic = topic

propan/brokers/rabbit/rabbit_broker.py

@@ -330,6 +330,7 @@ async def wrapper(message: RabbitMessage) -> T:
 
             async with context:
                 r = await func(message)
+
                 if message.reply_to:
                     await self.publish(
                         message=r,

propan/test/mqtt.py

@@ -0,0 +1,69 @@
+import sys
+from types import MethodType
+from typing import Any, Dict, Optional, Union
+
+from typing_extensions import TypeAlias
+
+if sys.version_info < (3, 8):
+    from asyncmock import AsyncMock
+else:
+    from unittest.mock import AsyncMock
+
+from paho.mqtt.client import MQTTMessage
+
+from propan._compat import model_to_json
+from propan.brokers.mqtt.mqtt_broker import MqttBroker
+from propan.brokers.mqtt.schemas import Handler
+from propan.test.utils import call_handler
+from propan.types import SendableMessage
+
+__all__ = (
+    "build_message",
+    "TestMqttBroker",
+)
+
+Msg: TypeAlias = Dict[str, Union[bytes, str, None]]
+
+
+def build_message(
+    message: SendableMessage,
+    topic: str,
+    *,
+    headers: Optional[Dict[str, Any]] = None,
+) -> Msg:
+    msg = MQTTMessage()
+    msg.topic = topic
+    msg.payload = message
+    return msg
+
+
+async def publish(
+    self: MqttBroker,
+    message: SendableMessage,
+    topic: str,
+    *,
+    reply_to: str = "",
+    headers: Optional[Dict[str, Any]] = None,
+    callback: bool = False,
+    callback_timeout: Optional[float] = 30.0,
+    raise_timeout: bool = False,
+) -> Any:
+    incoming = build_message(
+        message=message,
+        topic=topic,
+    )
+
+    for handler in self.handlers:  # pragma: no branch
+        if handler.topic == topic:
+            r = await call_handler(
+                handler, incoming, callback, callback_timeout, raise_timeout
+            )
+            if callback:  # pragma: no branch
+                return r
+
+
+def TestMqttBroker(broker: MqttBroker) -> MqttBroker:
+    broker.connect = AsyncMock()  # type: ignore
+    broker.start = AsyncMock()  # type: ignore
+    broker.publish = MethodType(publish, broker)  # type: ignore
+    return broker

pyproject.toml

@@ -43,7 +43,7 @@ classifiers = [
 ]
 
 dependencies = [
-    "fast-depends>=2.0.3",
+    "fast-depends>=2.1.1",
     "watchfiles",
     "typer",
     "uvloop>=0.14.0,!=0.15.0,!=0.15.1; sys_platform != 'win32' and (sys_platform != 'cygwin' and platform_python_implementation != 'PyPy')",
@@ -77,6 +77,14 @@ async-kafka = [
     "aiokafka>=0.8",
 ]
 
+async-sqs = [
+    "aiobotocore",
+]
+
+async-mqtt = [
+    "paho-mqtt>=1.6.1",
+]
+
 doc = [
     "PyYAML",
     "email-validator>=2.0.0",
@@ -86,10 +94,6 @@ doc = [
     "uvicorn",
 ]
 
-async-sqs = [
-    "aiobotocore",
-]
-
 testsuite = [
     "coverage[toml]>=7.2",
     "pytest==7.4.0",
@@ -107,6 +111,7 @@ test = [
     "propan[async-redis]",
     "propan[async-kafka]",
     "propan[async-sqs]",
+    "propan[async-mqtt]",
     "propan[testsuite]",
 
     "fastapi>=0.100.0b2",
@@ -168,6 +173,7 @@ features = [
   "async-nats",
   "async-rabbit",
   "async-redis",
+  "async-mqtt",
 ]
 
 [[tool.hatch.envs.test.matrix]]
@@ -186,6 +192,7 @@ features = [
   "async-nats",
   "async-rabbit",
   "async-redis",
+  "async-mqtt",
 ]
 
 [tool.hatch.envs.test-last.scripts]
@@ -252,6 +259,7 @@ markers = [
     "redis",
     "kafka",
     "sqs",
+    "mqtt",
     "slow",
     "all",
     "run",