added reconnect_delay, reconnect_limit, and backups option to Connection

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@904634 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 91 insertions, 21 deletions

diff --git a/python/examples/api/drain b/python/examples/api/drain
index 485985f16d..ef1f050c8c 100755
--- a/python/examples/api/drain
+++ b/python/examples/api/drain
@@ -21,18 +21,32 @@
 import optparse
 from qpid.messaging import *
 from qpid.util import URL
+from qpid.log import enable, DEBUG, WARN
 
 parser = optparse.OptionParser(usage="usage: %prog [options] ADDRESS ...",
                                description="Drain messages from the supplied address.")
 parser.add_option("-b", "--broker", default="localhost",
                   help="connect to specified BROKER (default %default)")
+parser.add_option("-r", "--reconnect", action="store_true",
+                  help="enable auto reconnect")
+parser.add_option("-d", "--reconnect-delay", type=float, default=3,
+                  help="delay between reconnect attempts")
+parser.add_option("-l", "--reconnect-limit", type=int,
+                  help="maximum number of reconnect attempts")
 parser.add_option("-t", "--timeout", type=float, default=0,
                   help="timeout in seconds to wait before exiting (default %default)")
 parser.add_option("-f", "--forever", action="store_true",
                   help="ignore timeout and wait forever")
+parser.add_option("-v", dest="verbose", action="store_true",
+                  help="enable logging")
 
 opts, args = parser.parse_args()
 
+if opts.verbose:
+  enable("qpid", DEBUG)
+else:
+  enable("qpid", WARN)
+
 url = URL(opts.broker)
 if args:
   addr = args.pop(0)
@@ -45,7 +59,11 @@ else:
 
 # XXX: should make URL default the port for us
 conn = Connection.open(url.host, url.port or AMQP_PORT,
-                       username=url.user, password=url.password)
+                       username=url.user,
+                       password=url.password,
+                       reconnect=opts.reconnect,
+                       reconnect_delay=opts.reconnect_delay,
+                       reconnect_limit=opts.reconnect_limit)
 ssn = conn.session()
 rcv = ssn.receiver(addr)
 
diff --git a/python/examples/api/server b/python/examples/api/server
index adb2dcf792..a9cd8579e3 100755
--- a/python/examples/api/server
+++ b/python/examples/api/server
@@ -28,7 +28,14 @@ parser = optparse.OptionParser(usage="usage: %prog [options] ADDRESS ...",
                                description="handle requests from the supplied address.")
 parser.add_option("-b", "--broker", default="localhost",
                   help="connect to specified BROKER (default %default)")
-parser.add_option("-v", dest="verbose", action="store_true", help="enable logging")
+parser.add_option("-r", "--reconnect", action="store_true",
+                  help="enable auto reconnect")
+parser.add_option("-d", "--reconnect-delay", type=float, default=3,
+                  help="delay between reconnect attempts")
+parser.add_option("-l", "--reconnect-limit", type=int,
+                  help="maximum number of reconnect attempts")
+parser.add_option("-v", dest="verbose", action="store_true",
+                  help="enable logging")
 
 opts, args = parser.parse_args()
 
@@ -45,8 +52,11 @@ else:
 
 # XXX: should make URL default the port for us
 conn = Connection.open(url.host, url.port or AMQP_PORT,
-                       username=url.user, password=url.password)
-conn.reconnect = True
+                       username=url.user,
+                       password=url.password,
+                       reconnect=opts.reconnect,
+                       reconnect_delay=opts.reconnect_delay,
+                       reconnect_limit=opts.reconnect_limit)
 ssn = conn.session()
 rcv = ssn.receiver(addr)
 
diff --git a/python/examples/api/spout b/python/examples/api/spout
index 1928303e43..ad98c486fd 100755
--- a/python/examples/api/spout
+++ b/python/examples/api/spout
@@ -21,6 +21,7 @@
 import optparse, time
 from qpid.messaging import *
 from qpid.util import URL
+from qpid.log import enable, DEBUG, WARN
 
 def nameval(st):
   idx = st.find("=")
@@ -36,20 +37,33 @@ parser = optparse.OptionParser(usage="usage: %prog [options] ADDRESS [ CONTENT .
                                description="Send messages to the supplied address.")
 parser.add_option("-b", "--broker", default="localhost",
                   help="connect to specified BROKER (default %default)")
+parser.add_option("-r", "--reconnect", action="store_true",
+                  help="enable auto reconnect")
+parser.add_option("-d", "--reconnect-delay", type=float, default=3,
+                  help="delay between reconnect attempts")
+parser.add_option("-l", "--reconnect-limit", type=int,
+                  help="maximum number of reconnect attempts")
 parser.add_option("-c", "--count", type=int, default=1,
                   help="stop after count messages have been sent, zero disables (default %default)")
 parser.add_option("-t", "--timeout", type=float, default=None,
                   help="exit after the specified time")
 parser.add_option("-i", "--id", help="use the supplied id instead of generating one")
-parser.add_option("-r", "--reply-to", help="specify reply-to address")
+parser.add_option("-R", "--reply-to", help="specify reply-to address")
 parser.add_option("-P", "--property", dest="properties", action="append", default=[],
                   metavar="NAME=VALUE", help="specify message property")
 parser.add_option("-M", "--map", dest="entries", action="append", default=[],
                   metavar="KEY=VALUE",
                   help="specify map entry for message body")
+parser.add_option("-v", dest="verbose", action="store_true",
+                  help="enable logging")
 
 opts, args = parser.parse_args()
 
+if opts.verbose:
+  enable("qpid", DEBUG)
+else:
+  enable("qpid", WARN)
+
 url = URL(opts.broker)
 if opts.id is None:
   spout_id = str(uuid4())
@@ -79,7 +93,11 @@ else:
 
 # XXX: should make URL default the port for us
 conn = Connection.open(url.host, url.port or AMQP_PORT,
-                       username=url.user, password=url.password)
+                       username=url.user,
+                       password=url.password,
+                       reconnect=opts.reconnect,
+                       reconnect_delay=opts.reconnect_delay,
+                       reconnect_limit=opts.reconnect_limit)
 ssn = conn.session()
 snd = ssn.sender(addr)
 
diff --git a/python/qpid/driver.py b/python/qpid/driver.py
index 8a1776d72c..03ba85aa39 100644
--- a/python/qpid/driver.py
+++ b/python/qpid/driver.py
@@ -58,7 +58,8 @@ DURABLE_DEFAULT=True
 # XXX
 
 FILTER_DEFAULTS = {
-  "topic": Pattern("*")
+  "topic": Pattern("*"),
+  "amq.failover": Pattern("DUMMY")
   }
 
 # XXX
@@ -148,6 +149,11 @@ class Driver:
     self._lock = self.connection._lock
 
     self._selector = Selector.default()
+    self._attempts = 0
+    self._hosts = [(self.connection.host, self.connection.port)] + \
+        self.connection.backups
+    self._host = 0
+    self._retrying = False
     self.reset()
 
   def reset(self):
@@ -286,7 +292,6 @@ class Driver:
 
   @synchronized
   def timeout(self):
-    log.warn("retrying ...")
     self.dispatch()
     self.connection._waiter.notifyAll()
 
@@ -294,10 +299,19 @@ class Driver:
     if self._socket is not None:
       self._socket.close()
     self.reset()
-    if recoverable and self.connection.reconnect:
-      self._timeout = time.time() + 3
-      log.warn("recoverable error: %s" % err)
-      log.warn("sleeping 3 seconds")
+    if (recoverable and self.connection.reconnect and
+        (self.connection.reconnect_limit is None or
+         self.connection.reconnect_limit <= 0 or
+         self._attempts <= self.connection.reconnect_limit)):
+      if self._host > 0:
+        delay = 0
+      else:
+        delay = self.connection.reconnect_delay
+      self._timeout = time.time() + delay
+      log.warn("recoverable error[attempt %s]: %s" % (self._attempts, err))
+      if delay > 0:
+        log.warn("sleeping %s seconds" % delay)
+      self._retrying = True
     else:
       self.connection.error = (err,)
 
@@ -430,16 +444,23 @@ class Driver:
   def connect(self):
     try:
       # XXX: should make this non blocking
-      self._socket = connect(self.connection.host, self.connection.port)
+      if self._host == 0:
+        self._attempts += 1
+      host, port = self._hosts[self._host]
+      if self._retrying:
+        log.warn("trying: %s:%s", host, port)
+      self._socket = connect(host, port)
+      if self._retrying:
+        log.warn("reconnect succeeded: %s:%s", host, port)
       self._timeout = None
+      self._attempts = 0
+      self._host = 0
+      self._retrying = False
+      self._buf += struct.pack(HEADER, "AMQP", 1, 1, 0, 10)
+      self._opening = True
     except socket.error, e:
-      if self.connection.reconnect:
-        self._error(e, True)
-        return
-      else:
-        raise e
-    self._buf += struct.pack(HEADER, "AMQP", 1, 1, 0, 10)
-    self._opening = True
+      self._host = (self._host + 1) % len(self._hosts)
+      self._error(e, True)
 
   def disconnect(self):
     self.write_op(ConnectionClose(close_code.normal))
diff --git a/python/qpid/messaging.py b/python/qpid/messaging.py
index 91d7bca703..9108832eec 100644
--- a/python/qpid/messaging.py
+++ b/python/qpid/messaging.py
@@ -108,12 +108,15 @@ class Connection:
     self.password = password
     self.mechanisms = options.get("mechanisms")
     self.heartbeat = options.get("heartbeat")
+    self.reconnect = options.get("reconnect", False)
+    self.reconnect_delay = options.get("reconnect_delay", 3)
+    self.reconnect_limit = options.get("reconnect_limit")
+    self.backups = options.get("backups", [])
     self.options = options
 
     self.id = str(uuid4())
     self.session_counter = 0
     self.sessions = {}
-    self.reconnect = options.get("reconnect", False)
     self._connected = False
     self._lock = RLock()
     self._condition = Condition(self._lock)