On 15/08/16 15:52, Chenxiong Qi wrote:
Hi all,
I searched a lot to see if qpid.proton really does not support
failover that I'm told. When I'm looking back to qpid.proton
documentation, I find Container.connect method, there is a argument
urls that accepts a sequence of broker urls. Is this for the
connection failover?
Yes. The 'reconnect' named name argument to connect will control whether
the library will automatically try to reconnect if the underlying tcp
socket is disconnected.
If you want to try different urls when reconnecting, you can specify a
list of them.
Meanwhile, I don't find any document and examples
about how to use Container.connect, could you give some explanation?
There is a brief tutorial for the Connector:
https://qpid.apache.org/releases/qpid-proton-0.13.1/proton/python/book/tutorial.html
along with a short overview of the key classes and methods:
https://qpid.apache.org/releases/qpid-proton-0.13.1/proton/python/book/overview.html
The examples from the tutorial are available in git (or are installed
with your proton install):
https://git1-us-west.apache.org/repos/asf/qpid-proton/repo?p=qpid-proton.git;a=tree;f=examples/python;hb=HEAD
However none of them actual show using a list of urls. I've attached one
that does. To use it, specify multiple servers to connect to using the
--address option, e.g.
failover.py --address host1 --address host2
or
failover.py --address localhost:5673 --address localhost:5674
It would be much appreciated if you can give an example. If it's for
failover, can it used to connect ActiveMQ in order to utilize ActiveMQ
Failover http://activemq.apache.org/failover-transport-reference.html
?
There is no built in support for retrieving urls to try from any broker.
However if you provide a list of urls, it will use them when reconnecting.
#!/usr/bin/env python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
from __future__ import print_function, unicode_literals
import optparse
from proton import Message
from proton.handlers import MessagingHandler
from proton.reactor import Container
class Client(MessagingHandler):
def __init__(self, urls, node, count):
super(Client, self).__init__()
self.urls = urls or ["localhost:5672"]
self.node = node
self.received = 0
self.sent = 0
self.confirmed = 0
self.total = count
def on_start(self, event):
conn = event.container.connect(urls=self.urls)
event.container.create_receiver(conn, self.node)
event.container.create_sender(conn, self.node)
def on_message(self, event):
if event.message.id and event.message.id < self.received:
# ignore duplicate message
return
if self.total == 0 or self.received < self.total:
self.received += 1
if self.received == self.total:
event.receiver.close()
event.connection.close()
elif (self.received % 1000) == 0:
print("received %s" % event.message.body)
def on_sendable(self, event):
while event.sender.credit and (self.total == 0 or self.sent < self.total):
msg = Message(id=(self.sent+1), body='message-%s' % (self.sent+1))
event.sender.send(msg)
self.sent += 1
def on_accepted(self, event):
self.confirmed += 1
if self.confirmed == self.total:
print("all messages confirmed")
event.connection.close()
def on_disconnected(self, event):
self.sent = self.confirmed
parser = optparse.OptionParser(usage="usage: %prog [options]")
parser.add_option("-a", "--address", action="append", default=[],
help="url(s) of servers to connect to (default %default)")
parser.add_option("-n", "--node", default="examples",
help="AMQP node to send tp and receive from e.g. a queue name (default %default)")
parser.add_option("-m", "--messages", type="int", default=0,
help="number of messages to send; 0 keeps sending indefinitely (default %default)")
opts, args = parser.parse_args()
try:
Container(Client(opts.address, opts.node, opts.messages)).run()
except KeyboardInterrupt: pass
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
