openstack/oslo.messaging
Code Issues Proposed changes

Port the AMQP 1.0 driver to Python 3

Browse Source 
Minor syntax fixes to allow the AMQP 1.0 driver to run under Python
2.7 and 3.4.

Closes-Bug: #1477124
Change-Id: I8512bb8274fad78285ab59797ba9b3f0f8b5c171
This commit is contained in:
Kenneth Giusti 2015-07-22 13:09:54 -04:00 committed by Victor Stinner
parent 29dc193179
commit 3419c95207
3 changed files with 11 additions and 7 deletions
oslo_messaging
_drivers/protocols/amqp
controller.pyeventloop.py
tests
test_amqp_driver.py

12
oslo_messaging/_drivers/protocols/amqp/controller.py

@ -578,7 +578,7 @@ class Controller(pyngus.ConnectionEventHandler):
""" """
LOG.debug("Connection active (%s:%i), subscribing...", LOG.debug("Connection active (%s:%i), subscribing...",
self.hosts.current.hostname, self.hosts.current.port) self.hosts.current.hostname, self.hosts.current.port)
for s in self._servers.itervalues(): for s in self._servers.values():
s.attach(self._socket_connection.connection) s.attach(self._socket_connection.connection)
self._replies = Replies(self._socket_connection.connection, self._replies = Replies(self._socket_connection.connection,
lambda: self._reply_link_ready()) lambda: self._reply_link_ready())
@ -641,10 +641,12 @@ class Controller(pyngus.ConnectionEventHandler):
if not self._reconnecting: if not self._reconnecting:
self._reconnecting = True self._reconnecting = True
self._replies = None self._replies = None
d = self._delay LOG.info("delaying reconnect attempt for %d seconds",
LOG.info("delaying reconnect attempt for %d seconds", d) self._delay)
self.processor.schedule(lambda: self._do_reconnect(), d) self.processor.schedule(lambda: self._do_reconnect(),
self._delay = 1 if self._delay == 0 else min(d * 2, 60) self._delay)
self._delay = (1 if self._delay == 0
else min(self._delay * 2, 60))
def _do_reconnect(self): def _do_reconnect(self):
"""Invoked on connection/socket failure, failover and re-connect to the """Invoked on connection/socket failure, failover and re-connect to the

2
oslo_messaging/_drivers/protocols/amqp/eventloop.py

@ -209,7 +209,7 @@ class Requests(object):
""" """
if request: if request:
self._requests.put(request) self._requests.put(request)
os.write(self._wakeup_pipe[1], "!") os.write(self._wakeup_pipe[1], b'!')
def fileno(self): def fileno(self):
"""Allows this request queue to be used by select().""" """Allows this request queue to be used by select()."""

4
oslo_messaging/tests/test_amqp_driver.py

@ -383,6 +383,8 @@ class TestCyrusAuthentication(test_utils.BaseTestCase):
# test_authentication_bad_mechs test below # test_authentication_bad_mechs test below
mechs = "DIGEST-MD5 SCRAM-SHA-1 CRAM-MD5 PLAIN" mechs = "DIGEST-MD5 SCRAM-SHA-1 CRAM-MD5 PLAIN"
t = Template("""sasldb_path: ${db} t = Template("""sasldb_path: ${db}
pwcheck_method: auxprop
auxprop_plugin: sasldb
mech_list: ${mechs} mech_list: ${mechs}
""") """)
with open(conf, 'w') as f: with open(conf, 'w') as f:
@ -806,7 +808,7 @@ class FakeBroker(threading.Thread):
"""Shutdown the server.""" """Shutdown the server."""
LOG.debug("Stopping test Broker %s:%d", self.host, self.port) LOG.debug("Stopping test Broker %s:%d", self.host, self.port)
self._shutdown = True self._shutdown = True
os.write(self._wakeup_pipe[1], "!") os.write(self._wakeup_pipe[1], b'!')
self.join() self.join()
LOG.debug("Test Broker %s:%d stopped", self.host, self.port) LOG.debug("Test Broker %s:%d stopped", self.host, self.port)