3
\] @ sV d dl Z d dlZy d dlZW n ek
r4 dZY nX ddlmZ G dd deZdS ) N )
PubSubManagerc sj e Zd ZdZdZd fdd Z fd d
Zd d Zd
d Zdd Z dd Z
dd Z dd Z dd Z
ZS ) KombuManagera Client manager that uses kombu for inter-process messaging.
This class implements a client manager backend for event sharing across
multiple processes, using RabbitMQ, Redis or any other messaging mechanism
supported by `kombu `_.
To use a kombu backend, initialize the :class:`Server` instance as
follows::
url = 'amqp://user:password@hostname:port//'
server = socketio.Server(client_manager=socketio.KombuManager(url))
:param url: The connection URL for the backend messaging queue. Example
connection URLs are ``'amqp://guest:guest@localhost:5672//'``
and ``'redis://localhost:6379/'`` for RabbitMQ and Redis
respectively. Consult the `kombu documentation
`_ for more on how to construct
connection URLs.
:param channel: The channel name on which the server sends and receives
notifications. Must be the same in all the servers.
:param write_only: If set ot ``True``, only initialize to emit events. The
default of ``False`` initializes the class for emitting
and receiving.
:param connection_options: additional keyword arguments to be passed to
``kombu.Connection()``.
:param exchange_options: additional keyword arguments to be passed to
``kombu.Exchange()``.
:param queue_options: additional keyword arguments to be passed to
``kombu.Queue()``.
:param producer_options: additional keyword arguments to be passed to
``kombu.Producer()``.
kombu#amqp://guest:guest@localhost:5672//socketioFNc sb t d krtdtt| j|||d || _|p2i | _|p dt tj }dddid}|j| j tj|| j f|S )Nzflask-socketio.Fz x-expiresi )r! Zqueue_arguments)struuiduuid4r" r r Queuer$ )r Z
queue_namer# r r r _queueY s zKombuManager._queuec C s | j jf d| j i| jS )NZexchange)r ZProducerr$ r )r r r r r _ s zKombuManager._producerc C s | j jdj| d S )Nz Sleeping {}s) _get_logger exceptionformat)r r+ intervalr r r Z__error_callbackc s z KombuManager.__error_callbackc C s2 | j }|j| j| jj| jd}|tj| d S )N)errback)r Zensurer publish _KombuManager__error_callbackpickledumps)r data
connectionr/ r r r _publishf s
zKombuManager._publishc c s | j }xx| j j| jd}y<|j|(}x |jdd}|j |jV q,W W d Q R X W q
|jk
r| | j j
d Y q
X q
W d S )N)r. T)blockz)Connection error while reading from queue) r) r Zensure_connectionr0 Z SimpleQueuegetZackpayloadZconnection_errorsr* r+ )r Z reader_queuer4 queuemessager r r _listenl s
zKombuManager._listen)r r FNNNNN)__name__
__module__ __qualname____doc__namer
r r r$ r) r r0 r5 r;
__classcell__r r )r r r s ! r )r1 r& r ImportErrorZpubsub_managerr r r r r r s