/usr/lib/python2.7/dist-packages/pika/adapters/tornado_connection.py is in python-pika 0.10.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 | """Use pika with the Tornado IOLoop"""
from tornado import ioloop
import logging
import time
from pika.adapters import base_connection
LOGGER = logging.getLogger(__name__)
class TornadoConnection(base_connection.BaseConnection):
"""The TornadoConnection runs on the Tornado IOLoop. If you're running the
connection in a web app, make sure you set stop_ioloop_on_close to False,
which is the default behavior for this adapter, otherwise the web app
will stop taking requests.
: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 bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
:param custom_ioloop: Override using the global IOLoop in Tornado
"""
WARN_ABOUT_IOLOOP = True
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 TornadoConnection 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 bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
:param custom_ioloop: Override using the global IOLoop in Tornado
"""
self.sleep_counter = 0
self.ioloop = custom_ioloop or ioloop.IOLoop.instance()
super(TornadoConnection, self).__init__(parameters, on_open_callback,
on_open_error_callback,
on_close_callback, self.ioloop,
stop_ioloop_on_close)
def _adapter_connect(self):
"""Connect to the remote socket, adding the socket to the IOLoop if
connected.
:rtype: bool
"""
error = super(TornadoConnection, self)._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(TornadoConnection, self)._adapter_disconnect()
def add_timeout(self, deadline, callback_method):
"""Add the callback_method to the IOLoop timer to fire after deadline
seconds. Returns a handle to the timeout. Do not confuse with
Tornado's timeout where you pass in the time you want to have your
callback called. Only pass in the seconds until it's to be called.
:param int deadline: The number of seconds to wait to call callback
:param method callback_method: The callback method
:rtype: str
"""
return self.ioloop.add_timeout(time.time() + deadline, callback_method)
def remove_timeout(self, timeout_id):
"""Remove the timeout from the IOLoop by the ID returned from
add_timeout.
:rtype: str
"""
return self.ioloop.remove_timeout(timeout_id)
|