You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2017/07/06 15:36:22 UTC
qpid-cpp git commit: QPID-7847: qmf.client.BrokerAgent does not
provide reconnect support
Repository: qpid-cpp
Updated Branches:
refs/heads/master e7ab5476c -> 0ffa89e17
QPID-7847: qmf.client.BrokerAgent does not provide reconnect support
Provided via BrokerAgent.connect(), for example:
BrokerAgent.connect(self.server.addr, reconnect_delays=ReconnectDelays(.125, 1, True)
After a connection failure this will try to reconnect with the following delays (seconds)
0, .125, .25, .5, 1, 1, 1, 1 ...
You can use any iterable as the reconnect_delays, ReconnectDelays provides a convenience
to generate an exponential (binary) backoff between two values, followed by a
fixed (or unlimited) repeat of the final value.
Project: http://git-wip-us.apache.org/repos/asf/qpid-cpp/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-cpp/commit/0ffa89e1
Tree: http://git-wip-us.apache.org/repos/asf/qpid-cpp/tree/0ffa89e1
Diff: http://git-wip-us.apache.org/repos/asf/qpid-cpp/diff/0ffa89e1
Branch: refs/heads/master
Commit: 0ffa89e177346e68bdeb5a24b9f25190c983bd34
Parents: e7ab547
Author: Alan Conway <ac...@redhat.com>
Authored: Thu Jul 6 11:32:03 2017 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Thu Jul 6 11:32:03 2017 -0400
----------------------------------------------------------------------
management/python/lib/qmf/client.py | 124 +++++++++++++++----
src/tests/qmf_client_tests.py | 199 ++++++++++++++++++++++++++++++-
2 files changed, 296 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-cpp/blob/0ffa89e1/management/python/lib/qmf/client.py
----------------------------------------------------------------------
diff --git a/management/python/lib/qmf/client.py b/management/python/lib/qmf/client.py
index bc8fe73..3ba8e4b 100644
--- a/management/python/lib/qmf/client.py
+++ b/management/python/lib/qmf/client.py
@@ -27,8 +27,46 @@ which makes it more suitable for environments that may fork.
"""
from proton import Message
-from proton.utils import BlockingConnection, IncomingMessageHandler
-import threading, struct
+from proton.utils import BlockingConnection, IncomingMessageHandler, ConnectionException
+import threading, struct, time
+
+class ReconnectDelays(object):
+ """
+ An iterable that returns a (possibly unlimited) sequence of delays to for
+ successive reconnect attempts.
+ """
+
+ def __init__(self, shortest, longest, repeat=True):
+ """
+ First delay is 0, then `shortest`. Successive delays are doubled up to a
+ maximum of `longest`. If `repeat` is a number > 0 then the `longest` value
+ is generated `repeat` more times. If `repeat` is True, the longest
+ value is returned without limit.
+
+ For example: ReconnectDelays(.125, 1, True) will generate the following sequence:
+ 0, .125, .25, .5, 1, 1, 1, 1, 1, 1 ... (forever)
+ """
+ if shortest <= 0 or shortest > longest or repeat < 0:
+ raise ValueError("invalid arguments for reconnect_delays()")
+ self.shortest, self.longest, self.repeat = shortest, longest, repeat
+
+ def _generate(self):
+ yield 0
+ delay = self.shortest
+ while delay < self.longest:
+ yield delay
+ delay *= 2
+ yield self.longest
+ if self.repeat is True:
+ while True:
+ yield self.longest
+ elif self.repeat:
+ for i in xrange(self.repeat):
+ yield self.longest
+
+ def __iter__(self):
+ return self._generate()
+
class SyncRequestResponse(IncomingMessageHandler):
"""
@@ -48,15 +86,18 @@ class SyncRequestResponse(IncomingMessageHandler):
Sucessive messages may have different addresses.
"""
super(SyncRequestResponse, self).__init__()
- self.connection = connection
self.address = address
+ self.response = None
+ self._cid = 0
+ self.lock = threading.Lock()
+ self.reconnect(connection)
+
+ def reconnect(self, connection):
+ self.connection = connection
self.sender = self.connection.create_sender(self.address)
# dynamic=true generates a unique address dynamically for this receiver.
# credit=1 because we want to receive 1 response message initially.
self.receiver = self.connection.create_receiver(None, dynamic=True, credit=1, handler=self)
- self.response = None
- self._cid = 0
- self.lock = threading.Lock()
def _next(self):
"""Get the next correlation ID"""
@@ -109,6 +150,7 @@ class SyncRequestResponse(IncomingMessageHandler):
self.response = event.message
self.connection.container.yield_() # Wake up the wait() loop to handle the message.
+
class BrokerAgent(object):
"""Proxy for a manageable Qpid broker"""
@@ -123,9 +165,17 @@ class BrokerAgent(object):
password=str(sasl.password) if sasl else None)
@staticmethod
- def connect(url=None, timeout=10, ssl_domain=None, sasl=None):
- """Return a BrokerAgent connected with the given parameters"""
- return BrokerAgent(BrokerAgent.connection(url, timeout, ssl_domain, sasl))
+ def connect(url=None, timeout=10, ssl_domain=None, sasl=None, reconnect_delays=None):
+ """
+ Return a BrokerAgent connected with the given parameters.
+ @param reconnect_delays: iterable of delays for successive automatic re-connect attempts,
+ see class ReconnectDelays. If None there is no automatic re-connect
+ """
+ f = lambda: BrokerAgent.connection(url, timeout, ssl_domain, sasl)
+ ba = BrokerAgent(f())
+ ba.make_connection = f
+ ba.reconnect_delays = reconnect_delays or []
+ return ba
def __init__(self, connection):
"""
@@ -135,6 +185,10 @@ class BrokerAgent(object):
"""
path = connection.url.path or "qmf.default.direct"
self._client = SyncRequestResponse(connection, path)
+ self.reconnect_delays = None
+
+ def reconnect(self, connection):
+ self._client.reconnect(connection)
def close(self):
"""Shut down the node"""
@@ -145,6 +199,24 @@ class BrokerAgent(object):
def __repr__(self):
return "%s(%s)"%(self.__class__.__name__, self._client.connection.url)
+ def _reconnect(self):
+ for d in self.reconnect_delays:
+ time.sleep(d)
+ try:
+ self.reconnect(self.make_connection())
+ return True
+ except ConnectionException:
+ pass
+ return False
+
+ def _retry(self, f, *args, **kwargs):
+ while True:
+ try:
+ return f(*args, **kwargs)
+ except ConnectionException:
+ if not self._reconnect():
+ raise
+
def _request(self, opcode, content):
props = {'method' : 'request',
'qmf.opcode' : opcode,
@@ -158,7 +230,7 @@ class BrokerAgent(object):
content = {'_object_id' : {'_object_name' : addr},
'_method_name' : method,
'_arguments' : arguments or {}}
- response = self._request('_method_request', content)
+ response = self._retry(self._request, '_method_request', content)
if response.properties['qmf.opcode'] == '_exception':
raise Exception("management error: %r" % response.body['_values'])
if response.properties['qmf.opcode'] != '_method_response':
@@ -174,20 +246,24 @@ class BrokerAgent(object):
def _classQuery(self, class_name):
query = {'_what' : 'OBJECT', '_schema_id' : {'_class_name' : class_name}}
- response = self._request('_query_request', query)
- if response.properties['qmf.opcode'] != '_query_response':
- raise Exception("bad response")
- return self._gather(response)
+ def f():
+ response = self._request('_query_request', query)
+ if response.properties['qmf.opcode'] != '_query_response':
+ raise Exception("bad response")
+ return self._gather(response)
+ return self._retry(f)
def _nameQuery(self, object_id):
query = {'_what' : 'OBJECT', '_object_id' : {'_object_name' : object_id}}
- response = self._request('_query_request', query)
- if response.properties['qmf.opcode'] != '_query_response':
- raise Exception("bad response")
- items = self._gather(response)
- if len(items) == 1:
- return items[0]
- return None
+ def f():
+ response = self._request('_query_request', query)
+ if response.properties['qmf.opcode'] != '_query_response':
+ raise Exception("bad response")
+ items = self._gather(response)
+ if len(items) == 1:
+ return items[0]
+ return None
+ return self._retry(f)
def _getAll(self, cls):
return [cls(self, x) for x in self._classQuery(cls.__name__.lower())]
@@ -219,8 +295,8 @@ class BrokerAgent(object):
def getMemory(self): return self._getSingle(Memory)
def echo(self, sequence = 1, body = "Body"):
- """Request a response to test the path to the management broker"""
- return self._method('echo', {'sequence' : sequence, 'body' : body})
+ """Request a response to test the path to the management broker"""
+ return self._method('echo', {'sequence' : sequence, 'body' : body})
def queueMoveMessages(self, srcQueue, destQueue, qty):
"""Move messages from one queue to another"""
@@ -344,7 +420,7 @@ class BrokerAgent(object):
def list(self, _type):
"""List objects of the specified type"""
- return [i["_values"] for i in self._doClassQuery(_type.lower())]
+ return [i["_values"] for i in self._classQuery(_type.lower())]
def query(self, _type, oid):
"""Query the current state of an object"""
http://git-wip-us.apache.org/repos/asf/qpid-cpp/blob/0ffa89e1/src/tests/qmf_client_tests.py
----------------------------------------------------------------------
diff --git a/src/tests/qmf_client_tests.py b/src/tests/qmf_client_tests.py
index 3264d22..66ac1fa 100755
--- a/src/tests/qmf_client_tests.py
+++ b/src/tests/qmf_client_tests.py
@@ -20,7 +20,37 @@
# Runs QMF tests using the qmf.client API.
-import unittest, qmf.client, os
+
+import unittest, os, socket, time
+from qmf.client import SyncRequestResponse, BrokerAgent, ReconnectDelays
+from proton.handlers import MessagingHandler
+from proton.reactor import Container
+from proton.utils import BlockingConnection, ConnectionException
+from proton import Message, Event
+from threading import Thread
+from Queue import Queue, Empty, Full
+
+class TestPort(object):
+ """Get an unused port using bind(0) and SO_REUSEADDR and hold it till close()
+ Can be used as `with TestPort() as tp:` Provides tp.host, tp.port and tp.addr
+ (a "host:port" string)
+ """
+ def __init__(self):
+ self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ self.sock.bind(('127.0.0.1', 0)) # Testing exampless is local only
+ self.host, self.port = socket.getnameinfo(self.sock.getsockname(), 0)
+ self.addr = "%s:%s" % (self.host, self.port)
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ self.close()
+
+ def close(self):
+ self.sock.close()
+
class QmfClientTest(unittest.TestCase):
"""
@@ -32,7 +62,7 @@ class QmfClientTest(unittest.TestCase):
self.broker = config.broker
def setUp(self):
- self.agent = qmf.client.BrokerAgent.connect(self.broker)
+ self.agent = BrokerAgent.connect(self.broker)
def test_broker(self):
self.assertEqual(self.agent.getBroker().name, "amqp-broker")
@@ -89,11 +119,174 @@ class QmfClientTest(unittest.TestCase):
self.assertEqual('child', self.agent.getQueue('child').name)
else: # child
# Can't use the parent's connection.
- agent = qmf.client.BrokerAgent.connect(self.broker)
+ agent = BrokerAgent.connect(self.broker)
agent.delQueue('parent')
agent.addQueue('child')
os._exit(0) # Force exit, test framework will catch SystemExit
+
+class DisconnectServer(MessagingHandler, Thread):
+ """
+ Server that disconnects its clients to test automatic re-connect
+ """
+ def __init__(self, addr):
+ Thread.__init__(self)
+ MessagingHandler.__init__(self)
+
+ self.addr = addr
+ self.response = None # Response message
+ self.senders = {}
+ self.listening = False
+
+ self.disconnect = Queue(0) # Disconnect requests
+ self.disconnected = Queue(0) # Disconnects executed
+
+ # Start listener and server thread
+ self.container = Container(self)
+ self.container.start()
+ while not self.listening and self.container.process():
+ pass
+ self.start()
+
+ def run(self):
+ while self.container.process():
+ pass
+ self.container.stop()
+ self.container.process()
+
+ def stop(self):
+ self.container.stop()
+ self.join()
+
+ def on_start(self, event):
+ self.acceptor = event.container.listen(self.addr)
+ self.listening = True
+
+ def on_connection_bound(self, event):
+ # Turn off security
+ event.transport.require_auth(False);
+ event.transport.sasl().allowed_mechs("ANONYMOUS");
+ self.transport = event.transport
+
+ def check_disconnect(self, event):
+ try:
+ self.disconnect.get_nowait()
+ event.transport.close_head()
+ event.transport.close_tail()
+ self.disconnected.put(event.type)
+ return True
+ except Empty:
+ return False
+
+ def on_link_opening(self, event):
+ if event.link.is_sender:
+ if event.link.remote_source == "STOP":
+ self.reactor.stop()
+ if event.link.remote_source and event.link.remote_source.dynamic:
+ event.link.source.address = str(id(event.link))
+ self.senders[event.link.source.address] = event.link
+ else:
+ event.link.source.address = event.link.remote_source.address
+ else:
+ event.link.target.address = event.link.remote_target.address
+ event.link.flow(1)
+ self.check_disconnect(event)
+
+ def on_message(self, event):
+ if self.check_disconnect(event):
+ return
+ em = event.message
+ m = self.response or em
+ m.address = em.reply_to
+ m.correlation_id = em.correlation_id
+ self.senders[m.address].send(m)
+ event.link.flow(1)
+
+
+class ReconnectTests(unittest.TestCase):
+
+ def setUp(self):
+ with TestPort() as tp:
+ self.server = DisconnectServer(tp.addr)
+
+ def tearDown(self):
+ self.server.stop()
+
+ def test_reconnect_delays(self):
+ self.assertEquals([0, 1, 2, 4, 7, 7, 7, 7], list(ReconnectDelays(1, 7, 3)))
+ self.assertEquals([0, .2, .4, .8, 1.0], list(ReconnectDelays(.2, 1, 0)))
+ self.assertRaises(ValueError, ReconnectDelays, 0, 1)
+ self.assertRaises(ValueError, ReconnectDelays, 1, -1)
+ d = iter(ReconnectDelays(5, 5)) # 5's forever
+ self.assertEquals(0, d.next())
+ for x in xrange(100):
+ self.assertEquals(5, d.next())
+
+ query_response = Message(body=[],
+ properties={"method":"response", "qmf.agent":"broker",
+ "qmf.content":"_data", "qmf.opcode":"_query_response"})
+
+ method_response = Message(body={"_arguments":[]},
+ properties={"method":"response", "qmf.agent":"broker",
+ "qmf.content":"_data", "qmf.opcode":"_method_response"})
+
+ def test_reconnect_agent(self):
+ # Dummy response message
+ self.server.response = self.query_response
+
+ # Failure during initial connection should raise an exception, no reconnect
+ self.server.disconnect.put(True)
+ self.assertRaises(ConnectionException, BrokerAgent.connect, self.server.addr, reconnect_delays=[0, 0, 0])
+ self.assertEquals(Event.LINK_REMOTE_OPEN, self.server.disconnected.get())
+
+ agent = BrokerAgent.connect(self.server.addr, reconnect_delays=[0, 0, 0])
+ agent.getBroker() # Should work OK
+
+ self.server.disconnect.put(True) # Disconnect on message delivery
+ self.server.disconnect.put(True) # Disconnect first reconnect on link open
+ self.server.disconnect.put(True) # Disconnect second reconnect on link open
+ agent.getBroker()
+ self.assertEquals(Event.DELIVERY, self.server.disconnected.get())
+ self.assertEquals(Event.LINK_REMOTE_OPEN, self.server.disconnected.get())
+ self.assertEquals(Event.LINK_REMOTE_OPEN, self.server.disconnected.get())
+
+ # Try a healthy get
+ agent.getBroker()
+
+ self.server.disconnect.put(True)
+ agent.list("foo")
+ self.assertEquals(Event.DELIVERY, self.server.disconnected.get())
+
+ self.server.disconnect.put(True)
+ agent.getConnection("foo")
+ self.assertEquals(Event.DELIVERY, self.server.disconnected.get())
+
+ # Try a method call
+ self.server.response = self.method_response
+ self.server.disconnect.put(True)
+ agent.echo()
+ self.assertEquals(Event.DELIVERY, self.server.disconnected.get())
+
+ # We should give up after 4 disconnects
+ self.server.disconnect.put(True)
+ self.server.disconnect.put(True)
+ self.server.disconnect.put(True)
+ self.server.disconnect.put(True)
+ self.assertRaises(ConnectionException, agent.echo)
+
+ def test_reconnect_agent_delay(self):
+ self.server.response = self.query_response
+ agent = BrokerAgent.connect(self.server.addr, reconnect_delays=[0.1, 0.2])
+ def elapsed(f, *args, **kwargs):
+ t = time.time()
+ f(*args, **kwargs)
+ return time.time() - t
+ self.server.disconnect.put(True)
+ self.assertLess(0.1, elapsed(agent.getBroker))
+ self.server.disconnect.put(True)
+ self.server.disconnect.put(True)
+ self.assertLess(0.3, elapsed(agent.getBroker))
+
if __name__ == "__main__":
shutil.rmtree("brokertest.tmp", True)
os.execvp("qpid-python-test", ["qpid-python-test", "-m", "qmf_client_tests"])
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org