URL params passing to aiormq (#569)

* Fixes #567

.editorconfig

@@ -8,13 +8,11 @@ trim_trailing_whitespace = true
 
 [*.{py,yml}]
 indent_style = space
+max_line_length = 79
 
 [*.py]
 indent_size = 4
 
-[docs/**.py]
-max_line_length = 80
-
 [*.rst]
 indent_size = 3
 

aio_pika/abc.py

@@ -2,13 +2,15 @@
 import dataclasses
 import sys
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 from enum import Enum, IntEnum, unique
 from functools import singledispatch
 from types import TracebackType
 from typing import (
     Any, AsyncContextManager, AsyncIterable, Awaitable, Callable, Dict,
-    Generator, Iterator, Optional, Type, TypeVar, Union, overload,
+    Generator, Iterator, Mapping, Optional, Tuple, Type, TypeVar, Union,
+    overload,
 )
 
 
@@ -692,10 +694,29 @@ async def close(self, exc: Optional[aiormq.abc.ExceptionType]) -> Any:
             await self.close_callback.wait()
 
 
+@dataclass
+class ConnectionParameter:
+    name: str
+    parser: Callable[[str], Any]
+    default: Optional[str] = None
+    is_kwarg: bool = False
+
+    def parse(self, value: Optional[str]) -> Any:
+        if value is None:
+            return self.default
+        try:
+            return self.parser(value)
+        except ValueError:
+            return self.default
+
+
 class AbstractConnection(PoolInstance, ABC):
+    PARAMETERS: Tuple[ConnectionParameter, ...]
+
     close_callbacks: CallbackCollection
     connected: asyncio.Event
     transport: Optional[UnderlayConnection]
+    kwargs: Mapping[str, Any]
 
     @abstractmethod
     def __init__(
@@ -912,6 +933,7 @@ def _get_exchange_name_from_str(value: str) -> str:
     "CallbackType",
     "ChannelCloseCallback",
     "ConnectionCloseCallback",
+    "ConnectionParameter",
     "ConsumerTag",
     "DateType",
     "DeclarationResult",

aio_pika/connection.py

@@ -1,15 +1,16 @@
 import asyncio
 from ssl import SSLContext
 from types import TracebackType
-from typing import Any, Callable, Dict, Optional, Tuple, Type, TypeVar, Union
+from typing import Any, Dict, Optional, Tuple, Type, TypeVar, Union
 
 import aiormq.abc
+from aiormq.connection import parse_int
 from pamqp.common import FieldTable
 from yarl import URL
 
 from .abc import (
-    AbstractChannel, AbstractConnection, SSLOptions, TimeoutType,
-    UnderlayConnection,
+    AbstractChannel, AbstractConnection, ConnectionParameter, SSLOptions,
+    TimeoutType, UnderlayConnection,
 )
 from .channel import Channel
 from .exceptions import ConnectionClosed
@@ -25,7 +26,18 @@ class Connection(AbstractConnection):
     """ Connection abstraction """
 
     CHANNEL_CLASS: Type[Channel] = Channel
-    KWARGS_TYPES: Tuple[Tuple[str, Callable[[str], Any], str], ...] = ()
+    PARAMETERS: Tuple[ConnectionParameter, ...] = (
+        ConnectionParameter(
+            name="interleave",
+            parser=parse_int,
+            is_kwarg=True,
+        ),
+        ConnectionParameter(
+            name="happy_eyeballs_delay",
+            parser=float,
+            is_kwarg=True,
+        ),
+    )
 
     _closed: bool
 
@@ -44,10 +56,16 @@ async def close(
         self._closed = True
 
     @classmethod
-    def _parse_kwargs(cls, kwargs: Dict[str, Any]) -> Dict[str, Any]:
+    def _parse_parameters(cls, kwargs: Dict[str, Any]) -> Dict[str, Any]:
         result = {}
-        for key, parser, default in cls.KWARGS_TYPES:
-            result[key] = parser(kwargs.get(key, default))
+        for parameter in cls.PARAMETERS:
+            value = kwargs.get(parameter.name, parameter.default)
+
+            if parameter.is_kwarg and value is None:
+                # skip optional value
+                continue
+
+            result[parameter.name] = parameter.parse(value)
         return result
 
     def __init__(
@@ -61,7 +79,7 @@ def __init__(
 
         self.url = URL(url)
 
-        self.kwargs: Dict[str, Any] = self._parse_kwargs(
+        self.kwargs: Dict[str, Any] = self._parse_parameters(
             kwargs or dict(self.url.query),
         )
         self.kwargs["context"] = ssl_context
@@ -321,7 +339,7 @@ async def main():
 
         ``client_properties`` argument requires ``aiormq>=2.9``
 
-    URL string might be contain ssl parameters e.g.
+    URL string might be containing ssl parameters e.g.
     `amqps://user:pass@host//?ca_certs=ca.pem&certfile=crt.pem&keyfile=key.pem`
 
     :param client_properties: add custom client capability.
@@ -371,4 +389,4 @@ async def main():
     return connection
 
 
-__all__ = ("connect", "Connection", "make_url")
+__all__ = ("Connection", "connect", "make_url")

aio_pika/robust_connection.py

@@ -1,6 +1,6 @@
 import asyncio
 from ssl import SSLContext
-from typing import Any, Optional, Type, Union
+from typing import Any, Optional, Tuple, Type, Union
 from weakref import WeakSet
 
 import aiormq.abc
@@ -9,7 +9,8 @@
 from yarl import URL
 
 from .abc import (
-    AbstractRobustChannel, AbstractRobustConnection, SSLOptions, TimeoutType,
+    AbstractRobustChannel, AbstractRobustConnection, ConnectionParameter,
+    SSLOptions, TimeoutType,
 )
 from .connection import Connection, make_url
 from .exceptions import CONNECTION_EXCEPTIONS
@@ -26,9 +27,15 @@ class RobustConnection(Connection, AbstractRobustConnection):
 
     CHANNEL_REOPEN_PAUSE = 1
     CHANNEL_CLASS: Type[RobustChannel] = RobustChannel
-    KWARGS_TYPES = (
-        ("reconnect_interval", parse_timeout, "5"),
-        ("fail_fast", parse_bool, "1"),
+    PARAMETERS: Tuple[ConnectionParameter, ...] = Connection.PARAMETERS + (
+        ConnectionParameter(
+            name="reconnect_interval",
+            parser=parse_timeout, default="5",
+        ),
+        ConnectionParameter(
+            name="fail_fast",
+            parser=parse_bool, default="1",
+        ),
     )
 
     def __init__(
@@ -284,7 +291,7 @@ async def main():
 
         ``client_properties`` argument requires ``aiormq>=2.9``
 
-    URL string might be contain ssl parameters e.g.
+    URL string might contain ssl parameters e.g.
     `amqps://user:pass@host//?ca_certs=ca.pem&certfile=crt.pem&keyfile=key.pem`
 
     :param client_properties: add custom client capability.

docs/source/index.rst

@@ -52,6 +52,100 @@ Features
 .. _publisher confirms: https://www.rabbitmq.com/confirms.html
 .. _Transactions: https://www.rabbitmq.com/semantics.html#tx
 
+AMQP URL parameters
++++++++++++++++++++
+
+URL is the supported way to configure connection.
+For customisation of conneciton behaviour you might
+pass the parameters in URL query-string like format.
+
+This article describes a description for this parameters.
+
+``aiormq`` specific
+~~~~~~~~~~~~~~~~~~~
+
+* ``name`` (``str`` url encoded) - A string that will be visible in the RabbitMQ management console and in
+  the server logs, convenient for diagnostics.
+
+* ``cafile`` (``str``) - Path to Certificate Authority file
+
+* ``capath`` (``str``) - Path to Certificate Authority directory
+
+* ``cadata`` (``str`` url encoded) - URL encoded CA certificate content
+
+* ``keyfile`` (``str``) - Path to client ssl private key file
+
+* ``certfile`` (``str``) - Path to client ssl certificate file
+
+* ``no_verify_ssl`` - No verify server SSL certificates. ``0`` by default and means ``False`` other value means
+  ``True``.
+
+* ``heartbeat`` (``int``-like) - interval in seconds between AMQP heartbeat packets. ``0`` disables this feature.
+
+
+``aio_pika.connect`` function and ``aio_pika.Connection`` class specific
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+* ``interleave`` (``int``-like) - controls address reordering when a host name resolves to multiple
+  IP addresses. If 0 or unspecified, no reordering is done, and addresses are tried
+  in the order returned by ``getaddrinfo()``. If a positive integer is specified,
+  the addresses are interleaved by address family, and the given integer is interpreted
+  as "First Address Family Count" as defined in `RFC 8305`_. The default is ``0`` if
+  ``happy_eyeballs_delay`` is not specified, and ``1`` if it is.
+
+  .. note::
+
+      Really useful for RabbitMQ clusters with one DNS name with many ``A``/``AAAA`` records.
+
+  .. warning::
+
+      This option is supported by ``asyncio.DefaultEventLoopPolicy`` and available since python 3.8.
+
+* ``happy_eyeballs_delay`` (``float``-like) - if given, enables Happy Eyeballs for this connection.
+  It should be a floating-point number representing the amount of time in seconds to wait for a connection attempt
+  to complete, before starting the next attempt in parallel. This is the "Connection Attempt Delay" as defined in
+  `RFC 8305`_. A sensible default value recommended by the RFC is ``0.25`` (250 milliseconds).
+
+  .. note::
+
+      Really useful for RabbitMQ clusters with one DNS name with many ``A``/``AAAA`` records.
+
+  .. warning::
+
+      This option is supported by ``asyncio.DefaultEventLoopPolicy`` and available since python 3.8.
+
+``aio_pika.connect_robust`` function and ``aio_pika.RobustConnection`` class specific
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+For ``aio_pika.RobustConnection`` class is applicable all ``aio_pika.Connection`` related parameters like,
+``name``/``interleave``/``happy_eyeballs_delay`` and some specific:
+
+
+* ``reconnect_interval`` (``float``-like) - is the period in seconds, not more often than the attempts to
+  re-establish the connection will take place.
+
+
+* ``fail_fast`` (``true``/``yes``/``y``/``enable``/``on``/``enabled``/``1`` means ``True``, otherwise ``False``) -
+  special behavior for the start connection attempt, if it fails, all other attempts stops and an exception will be
+  thrown at the connection stage. Enabled by default, if you are sure you need to disable this feature, be ensures
+  for the passed URL is reallt working. Otherwise, your program will go into endless reconnection attempts that can
+  not be successed.
+
+.. _RFC 8305: https://datatracker.ietf.org/doc/html/rfc8305.html
+
+
+URL examples
+~~~~~~~~~~~~
+
+* ``amqp://username:password@hostname/vhost?name=connection%20name&heartbeat=60&happy_eyeballs_delay=0.25``
+
+* ``amqps://username:password@hostname/vhost?reconnect_interval=5&fail_fast=1``
+
+* ``amqps://username:password@hostname/vhost?cafile=/path/to/ca.pem``
+
+* ``amqps://username:password@hostname/vhost?cafile=/path/to/ca.pem&keyfile=/path/to/key.pem&certfile=/path/to/sert.pem``
+
+
 Installation
 ++++++++++++
 
@@ -283,7 +377,7 @@ The `patio`_ and the `patio-rabbitmq`_
    from patio_rabbitmq import RabbitMQBroker
 
    rpc = Registry(project="patio-rabbitmq", auto_naming=False)
-   
+
    @rpc("sum")
    def sum(*args):
        return sum(args)
@@ -325,10 +419,10 @@ If you need no deep dive into **RabbitMQ** details, you can use more high-level
 .. code-block:: python
 
    from propan import PropanApp, RabbitBroker
-   
+
    broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
    app = PropanApp(broker)
-   
+
    @broker.handle("user")
    async def user_created(user_id: int):
        assert isinstance(user_id, int)
@@ -355,15 +449,15 @@ Also this package is suitable for building messaging services over **RabbitMQ**
 
    import socketio
    from aiohttp import web
-   
+
    sio = socketio.AsyncServer(client_manager=socketio.AsyncAioPikaManager())
    app = web.Application()
    sio.attach(app)
-   
+
    @sio.event
    async def chat_message(sid, data):
        print("message ", data)
-   
+
    if __name__ == '__main__':
        web.run_app(app)
 
@@ -373,13 +467,13 @@ And a client is able to call `chat_message` the following way:
 
    import asyncio
    import socketio
-   
+
    sio = socketio.AsyncClient()
-   
+
    async def main():
        await sio.connect('http://localhost:8080')
        await sio.emit('chat_message', {'response': 'my response'})
-   
+
    if __name__ == '__main__':
        asyncio.run(main())
 
@@ -393,9 +487,9 @@ The library provides you with **aio-pika** broker for running tasks too.
 .. code-block:: python
 
    from taskiq_aio_pika import AioPikaBroker
-   
+
    broker = AioPikaBroker()
-   
+
    @broker.task
    async def test() -> None:
        print("nothing")
@@ -444,4 +538,4 @@ This software follows `Semantic Versioning`_
 .. _python-socketio: https://python-socketio.readthedocs.io/en/latest/intro.html
 .. _taskiq: https://github.com/taskiq-python/taskiq
 .. _taskiq-aio-pika: https://github.com/taskiq-python/taskiq-aio-pika
-.. _Rasa: https://rasa.com/docs/rasa/
+.. _Rasa: https://rasa.com/docs/rasa/