send

#!/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, time
from client import *

parser = optparse.OptionParser(usage="usage: %prog [options] <address> [<content> ...]",
                               description="send 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("-c", "--count", type=int, default=1,
                  help="number of messages to send (default %default)")
parser.add_option("-s", "--sleep", type=float,
                  help="sleep between sends for indicated period")
parser.add_option("-i", "--stdin", action="store_true",
                  help="read messages from stdin (one message per line)")
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 = Target(dynamic=True)
elif not args:
  parser.error("address is required")
else:
  if opts.durable:
    durable = 2
  else:
    durable = 0
  addr = Target(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()
lnk = ssn.sender(addr, name=opts.link)

if opts.dynamic:
  print lnk.address

count = 0
try:
  if opts.txn:
    txn = ssn.declare()
  else:
    txn = None

  while opts.count == 0 or count < opts.count:
    if opts.stdin:
      try:
        content = raw_input()
      except EOFError:
        break
    else:
      content = " ".join(args[1:])

    lnk.send(settled=False, message=Message(content, message_id=count), txn=txn)
    count += 1
    if opts.sleep:
      time.sleep(opts.sleep)

  if opts.txn:
    ssn.discharge(txn, fail=opts.rollback)

  while lnk.pending(True):
    for t, l, r in lnk.pending():
      print t, r.state
      lnk.settle(t, r.state)
except KeyboardInterrupt:
  pass

if opts.detach:
  lnk.detach()
else:
  lnk.close()
ssn.close()
conn.close()

for tag, local, remote in lnk.get_unsettled():
  print "UNSETTLED:", tag, local, remote