/usr/lib/python2.7/dist-packages/pika/adapters/asyncio_connection.py is in python-pika 0.11.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 | """Use pika with the Asyncio EventLoop"""
import asyncio
from functools import partial
from pika.adapters import base_connection
class IOLoopAdapter:
def __init__(self, loop):
"""
Basic adapter for asyncio event loop
:type loop: asyncio.AbstractEventLoop
:param loop: Asyncio Loop
"""
self.loop = loop
self.handlers = {}
self.readers = set()
self.writers = set()
def add_timeout(self, deadline, callback_method):
"""Add the callback_method to the EventLoop timer to fire after deadline
seconds. Returns a Handle to the timeout.
:param int deadline: The number of seconds to wait to call callback
:param method callback_method: The callback method
:rtype: asyncio.Handle
"""
return self.loop.call_later(deadline, callback_method)
@staticmethod
def remove_timeout(handle):
"""
Cancel asyncio.Handle
:type handle: asyncio.Handle
:rtype: bool
"""
return handle.cancel()
def add_handler(self, fd, cb, event_state):
""" Registers the given handler to receive the given events for ``fd``.
The ``fd`` argument is an integer file descriptor.
The ``event_state`` argument is a bitwise or of the constants
``base_connection.BaseConnection.READ``, ``base_connection.BaseConnection.WRITE``,
and ``base_connection.BaseConnection.ERROR``.
"""
if fd in self.handlers:
raise ValueError("fd {} added twice".format(fd))
self.handlers[fd] = cb
if event_state & base_connection.BaseConnection.READ:
self.loop.add_reader(
fd,
partial(
cb,
fd=fd,
events=base_connection.BaseConnection.READ
)
)
self.readers.add(fd)
if event_state & base_connection.BaseConnection.WRITE:
self.loop.add_writer(
fd,
partial(
cb,
fd=fd,
events=base_connection.BaseConnection.WRITE
)
)
self.writers.add(fd)
def remove_handler(self, fd):
""" Stop listening for events on ``fd``. """
if fd not in self.handlers:
return
if fd in self.readers:
self.loop.remove_reader(fd)
self.readers.remove(fd)
if fd in self.writers:
self.loop.remove_writer(fd)
self.writers.remove(fd)
del self.handlers[fd]
def update_handler(self, fd, event_state):
if event_state & base_connection.BaseConnection.READ:
if fd not in self.readers:
self.loop.add_reader(
fd,
partial(
self.handlers[fd],
fd=fd,
events=base_connection.BaseConnection.READ
)
)
self.readers.add(fd)
else:
if fd in self.readers:
self.loop.remove_reader(fd)
self.readers.remove(fd)
if event_state & base_connection.BaseConnection.WRITE:
if fd not in self.writers:
self.loop.add_writer(
fd,
partial(
self.handlers[fd],
fd=fd,
events=base_connection.BaseConnection.WRITE
)
)
self.writers.add(fd)
else:
if fd in self.writers:
self.loop.remove_writer(fd)
self.writers.remove(fd)
def start(self):
""" Start Event Loop """
if self.loop.is_running():
return
self.loop.run_forever()
def stop(self):
""" Stop Event Loop """
if self.loop.is_closed():
return
self.loop.stop()
class AsyncioConnection(base_connection.BaseConnection):
""" The AsyncioConnection runs on the Asyncio EventLoop.
:param pika.connection.Parameters parameters: Connection parameters
:param on_open_callback: The method to call when the connection is open
:type on_open_callback: method
:param on_open_error_callback: Method to call if the connection cant be opened
:type on_open_error_callback: method
:param asyncio.AbstractEventLoop loop: By default asyncio.get_event_loop()
"""
def __init__(self,
parameters=None,
on_open_callback=None,
on_open_error_callback=None,
on_close_callback=None,
stop_ioloop_on_close=False,
custom_ioloop=None):
""" Create a new instance of the AsyncioConnection class, connecting
to RabbitMQ automatically
:param pika.connection.Parameters parameters: Connection parameters
:param on_open_callback: The method to call when the connection is open
:type on_open_callback: method
:param on_open_error_callback: Method to call if the connection cant be opened
:type on_open_error_callback: method
:param asyncio.AbstractEventLoop loop: By default asyncio.get_event_loop()
"""
self.sleep_counter = 0
self.loop = custom_ioloop or asyncio.get_event_loop()
self.ioloop = IOLoopAdapter(self.loop)
super().__init__(
parameters, on_open_callback,
on_open_error_callback,
on_close_callback, self.ioloop,
stop_ioloop_on_close=stop_ioloop_on_close,
)
def _adapter_connect(self):
"""Connect to the remote socket, adding the socket to the EventLoop if
connected.
:rtype: bool
"""
error = super()._adapter_connect()
if not error:
self.ioloop.add_handler(
self.socket.fileno(),
self._handle_events,
self.event_state,
)
return error
def _adapter_disconnect(self):
"""Disconnect from the RabbitMQ broker"""
if self.socket:
self.ioloop.remove_handler(
self.socket.fileno()
)
super()._adapter_disconnect()
def _handle_disconnect(self):
# No other way to handle exceptions.ProbableAuthenticationError
try:
super()._handle_disconnect()
super()._handle_write()
except Exception as e:
# FIXME: Pass None or other constant instead "-1"
self._on_disconnect(-1, e)
|