Permalink
Switch branches/tags
Nothing to show
Find file
Fetching contributors…
Cannot retrieve contributors at this time
executable file 138 lines (124 sloc) 4.56 KB
#!/usr/bin/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.
#
import optparse, os, sys, time
from client import *
parser = optparse.OptionParser(usage="usage: %prog [options] <address>",
description="receive messages")
parser.add_option("-H", "--host",
help="host to connect to (default 0.0.0.0)")
parser.add_option("-a", "--auth", action="store_true",
help="enable sasl authentication layer")
parser.add_option("-u", "--username", help="username to use for authentication")
parser.add_option("-w", "--password", help="password to use for authentication")
parser.add_option("-m", "--mechanism", default="ANONYMOUS",
help="sasl mechanism to choose")
parser.add_option("-p", "--port", type=int, default=5672,
help="port to connect to (default %default)")
parser.add_option("-f", "--frame-size", type=int, default=4294967295,
help="specify the maximum frame size")
parser.add_option("-l", "--link", help="link name")
parser.add_option("-d", "--detach", action="store_true",
help="detach rather than close link")
parser.add_option("-D", "--dynamic", action="store_true",
help="set dynamic flag on terminus")
parser.add_option("-b", "--block", action="store_true",
help="block until messages arrive")
parser.add_option("-c", "--count", type=int, default=1,
help="number of messages to receive (default %default)")
parser.add_option("-s", "--sleep", type=float,
help="sleep between fetches for indicated period")
parser.add_option("-t", "--trace", default="err",
help="enable tracing for specified categories")
parser.add_option("-x", "--txn", action="store_true",
help="send transactionally")
parser.add_option("-r", "--rollback", action="store_true")
parser.add_option("-C", "--container", type=str,
help="specify the container-id")
parser.add_option("-e", "--durable", action="store_true",
help="specify terminus is dUrable")
opts, args = parser.parse_args()
if opts.dynamic:
addr = Source(dynamic=True)
elif not args:
parser.error("address is required")
else:
if opts.durable:
durable = 2
else:
durable = 0
addr = Source(address=args[0], durable=durable)
host = opts.host or os.getenv('AMQP_BROKER') or "0.0.0.0"
conn = Connection(auth=opts.auth)
conn.tracing(*opts.trace.split())
conn.connect(host, opts.port)
conn.open(mechanism=opts.mechanism.upper(), username=opts.username,
password=opts.password, max_frame_size=opts.frame_size,
container_id=opts.container)
ssn = conn.session()
if opts.count:
credit = min(10, opts.count)
else:
credit = 10
lnk = ssn.receiver(addr, limit=credit, drain=not opts.block, name=opts.link)
if opts.dynamic:
print lnk.address
count = 0
try:
if opts.txn:
txn = ssn.declare()
else:
txn = None
while lnk.pending(block=True):
msg = lnk.get()
count += 1
print msg
sys.stdout.flush()
if opts.sleep:
time.sleep(opts.sleep)
disp = ACCEPTED
if txn:
disp = TransactionalState(txn, disp)
lnk.disposition(msg.delivery_tag, disp)
if lnk.pending() < 10 and lnk.credit() == 0:
if opts.count == 0:
credit = 10
else:
credit = min(10, opts.count - (count + lnk.pending()))
if credit:
lnk.flow(credit, drain=not opts.block)
# XXX
for tag, l, _ in lnk.get_remote(settled=True):
if l.state is not None:
lnk.settle(tag)
if opts.txn:
ssn.discharge(txn, fail=opts.rollback)
except KeyboardInterrupt:
pass
if opts.detach:
lnk.detach()
else:
lnk.close()
ssn.close()
conn.close()
# XXX
for tag, _, _ in lnk.get_remote(settled=True):
lnk.settle(tag)
for tag, local, remote in lnk.get_unsettled():
print "UNSETTLED:", tag, local, remote