You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2010/01/29 22:41:46 UTC
svn commit: r904634 - in /qpid/trunk/qpid/python: examples/api/drain
examples/api/server examples/api/spout qpid/driver.py qpid/messaging.py
Author: rhs
Date: Fri Jan 29 21:41:46 2010
New Revision: 904634
URL: http://svn.apache.org/viewvc?rev=904634&view=rev
Log:
added reconnect_delay, reconnect_limit, and backups option to Connection
Modified:
qpid/trunk/qpid/python/examples/api/drain
qpid/trunk/qpid/python/examples/api/server
qpid/trunk/qpid/python/examples/api/spout
qpid/trunk/qpid/python/qpid/driver.py
qpid/trunk/qpid/python/qpid/messaging.py
Modified: qpid/trunk/qpid/python/examples/api/drain
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/python/examples/api/drain?rev=904634&r1=904633&r2=904634&view=diff
==============================================================================
--- qpid/trunk/qpid/python/examples/api/drain (original)
+++ qpid/trunk/qpid/python/examples/api/drain Fri Jan 29 21:41:46 2010
@@ -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 @@
# 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)
Modified: qpid/trunk/qpid/python/examples/api/server
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/python/examples/api/server?rev=904634&r1=904633&r2=904634&view=diff
==============================================================================
--- qpid/trunk/qpid/python/examples/api/server (original)
+++ qpid/trunk/qpid/python/examples/api/server Fri Jan 29 21:41:46 2010
@@ -28,7 +28,14 @@
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 @@
# 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)
Modified: qpid/trunk/qpid/python/examples/api/spout
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/python/examples/api/spout?rev=904634&r1=904633&r2=904634&view=diff
==============================================================================
--- qpid/trunk/qpid/python/examples/api/spout (original)
+++ qpid/trunk/qpid/python/examples/api/spout Fri Jan 29 21:41:46 2010
@@ -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 @@
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 @@
# 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)
Modified: qpid/trunk/qpid/python/qpid/driver.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/python/qpid/driver.py?rev=904634&r1=904633&r2=904634&view=diff
==============================================================================
--- qpid/trunk/qpid/python/qpid/driver.py (original)
+++ qpid/trunk/qpid/python/qpid/driver.py Fri Jan 29 21:41:46 2010
@@ -58,7 +58,8 @@
# XXX
FILTER_DEFAULTS = {
- "topic": Pattern("*")
+ "topic": Pattern("*"),
+ "amq.failover": Pattern("DUMMY")
}
# XXX
@@ -148,6 +149,11 @@
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 @@
@synchronized
def timeout(self):
- log.warn("retrying ...")
self.dispatch()
self.connection._waiter.notifyAll()
@@ -294,10 +299,19 @@
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 @@
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))
Modified: qpid/trunk/qpid/python/qpid/messaging.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/python/qpid/messaging.py?rev=904634&r1=904633&r2=904634&view=diff
==============================================================================
--- qpid/trunk/qpid/python/qpid/messaging.py (original)
+++ qpid/trunk/qpid/python/qpid/messaging.py Fri Jan 29 21:41:46 2010
@@ -108,12 +108,15 @@
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)
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org