@@ -0,0 +1,3462 @@
+import time%0A%0Aimport pika%0A%0A%0Aclass RobustConnection():%0A    def __init__(self, connection_parameters_kwargs, exchange_kwargs):%0A        self._connection_parameters_kwargs = connection_parameters_kwargs%0A        self._exchange_kwargs = exchange_kwargs%0A        self._connection = None%0A        self._channel = None%0A        self.connect()%0A        %0A    def __enter__(self):%0A        return self%0A    %0A    def __exit__(self, *args, **kwargs):%0A        self.disconnect()%0A    %0A    def _setup_connection(self):%0A        self._connection = pika.BlockingConnection(pika.ConnectionParameters(**self._connection_parameters_kwargs))%0A        %0A    def _setup_channel(self):%0A        self._channel = self._connection.channel()%0A    %0A    def _setup_exchange(self):%0A        self._channel.exchange_declare(**self._exchange_kwargs)%0A    %0A    def _connect(self):%0A        self._setup_connection()%0A        self._setup_channel()%0A        self._setup_exchange()%0A    %0A    def connect(self):%0A        if (self._connection is None) or self._connection.is_closed:%0A            self._connect()%0A        %0A    def disconnect(self):%0A        if self._connection is not None:%0A            self._connection.close()%0A        %0A    def retry(self, f, *args, **kwargs):%0A        wait_time = 0.25%0A        wait_time_max = 4.0%0A        while True:%0A            try:%0A                f(*args, **kwargs)%0A                break%0A            except (pika.exceptions.ConnectionClosedByBroker, pika.exceptions.AMQPConnectionError):%0A                time.sleep(wait_time)%0A                self._connect()%0A                wait_time = min(wait_time_max, 2.0 * wait_time)%0A                continue%0A        %0A%0Aclass Publisher(RobustConnection):%0A    def _publish(self, routing_key, body):%0A        self._channel.basic_publish(%0A            exchange=self._exchange_kwargs%5B%22exchange%22%5D,%0A            routing_key=routing_key,%0A            body=body,%0A        )%0A        %0A    def publish(self, *args, **kwargs):%0A        self.retry(f=self._publish, *args, **kwargs)%0A%0A%0Aclass Consumer(RobustConnection):%0A    def __init__(self, connection_parameters_kwargs, exchange_kwargs, queue_kwargs, bind_kwargs, consume_kwargss, prefetch_count=None):%0A        self._queue_kwargs = queue_kwargs%0A        self._bind_kwargs = bind_kwargs%0A        self._consume_kwargss = consume_kwargss%0A        self._prefetch_count = prefetch_count%0A        self._queue_name = None%0A        super().__init__(connection_parameters_kwargs=connection_parameters_kwargs, exchange_kwargs=exchange_kwargs)%0A%0A    def _setup_channel(self):%0A        super()._setup_channel()%0A        if self._prefetch_count is not None:%0A            self._channel.basic_qos(prefetch_count=self._prefetch_count)%0A%0A    def _setup_queue(self):%0A        result = self._channel.queue_declare(**self._queue_kwargs)%0A        self._queue_name = result.method.queue%0A        self._channel.queue_bind(**self._bind_kwargs, queue=self._queue_name, exchange=self._exchange_kwargs%5B%22exchange%22%5D, )%0A%0A    def _setup_consume(self):%0A        for consume_kwargs in self._consume_kwargss:%0A            if (%22queue%22 in consume_kwargs) and (consume_kwargs%5B%22queue%22%5D is None):%0A                consume_kwargs = self._queue_name%0A            self._channel.basic_consume(**consume_kwargs)%0A%0A    def _connect(self):%0A        super()._connect()%0A        self._setup_queue()%0A        self._setup_consume()%0A    %0A    def _start_consuming(self):%0A        self._channel.start_consuming()%0A        %0A    def start_consuming(self):%0A        self.retry(f=self._start_consuming)%0A%0A

Add first version of robust pika wrappers