diff options
Diffstat (limited to 'java/tools/bin')
| -rwxr-xr-x | java/tools/bin/Profile-run-from-source | 2 | ||||
| -rwxr-xr-x[-rw-r--r--] | java/tools/bin/mercury-controller | 0 | ||||
| -rwxr-xr-x[-rw-r--r--] | java/tools/bin/mercury-start-consumers | 0 | ||||
| -rwxr-xr-x[-rw-r--r--] | java/tools/bin/mercury-start-producers | 0 | ||||
| -rwxr-xr-x | java/tools/bin/qpid-jms-benchmark | 316 | ||||
| -rwxr-xr-x | java/tools/bin/qpid-jms-receive | 193 | ||||
| -rwxr-xr-x | java/tools/bin/qpid-jms-send | 261 |
7 files changed, 771 insertions, 1 deletions
diff --git a/java/tools/bin/Profile-run-from-source b/java/tools/bin/Profile-run-from-source index f8ec45ccff..179c365450 100755 --- a/java/tools/bin/Profile-run-from-source +++ b/java/tools/bin/Profile-run-from-source @@ -49,7 +49,7 @@ export LOG_CONFIG="-Dlog4j.configuration=file:///$QPID_CHECKOUT/java/tools/etc/t #------------- Required for qpid-python-testkit ----------------------------------------- -PYTHONPATH=$QPID_CHECKOUT/python/qpid:$QPID_CHECKOUT/cpp/src/test/brokertest.py:$PYTHONPATH +PYTHONPATH=$QPID_CHECKOUT/python:$QPID_CHECKOUT/cpp/src/test/brokertest.py:$PYTHONPATH export PATH=$QPID_CHECKOUT/python:$PATH if [ -x $QPID_CHECKOUT/cpp/src/qpidd ]; then diff --git a/java/tools/bin/mercury-controller b/java/tools/bin/mercury-controller index fab8614039..fab8614039 100644..100755 --- a/java/tools/bin/mercury-controller +++ b/java/tools/bin/mercury-controller diff --git a/java/tools/bin/mercury-start-consumers b/java/tools/bin/mercury-start-consumers index c71fc0c21f..c71fc0c21f 100644..100755 --- a/java/tools/bin/mercury-start-consumers +++ b/java/tools/bin/mercury-start-consumers diff --git a/java/tools/bin/mercury-start-producers b/java/tools/bin/mercury-start-producers index 7ba0286f7c..7ba0286f7c 100644..100755 --- a/java/tools/bin/mercury-start-producers +++ b/java/tools/bin/mercury-start-producers diff --git a/java/tools/bin/qpid-jms-benchmark b/java/tools/bin/qpid-jms-benchmark new file mode 100755 index 0000000000..3d712a27dc --- /dev/null +++ b/java/tools/bin/qpid-jms-benchmark @@ -0,0 +1,316 @@ +#!/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. +# + +import optparse, time, qpid.messaging, re +from threading import Thread +from subprocess import Popen, PIPE, STDOUT + +op = optparse.OptionParser(usage="usage: %prog [options]", + description="simple performance benchmarks") +op.add_option("-b", "--broker", default=[], action="append", type="str", + help="url of broker(s) to connect to, round robin on multiple brokers") +op.add_option("-c", "--client-host", default=[], action="append", type="str", + help="host(s) to run clients on via ssh, round robin on mulple hosts") +op.add_option("-q", "--queues", default=1, type="int", metavar="N", + help="create N queues (default %default)") +op.add_option("-s", "--senders", default=1, type="int", metavar="N", + help="start N senders per queue (default %default)") +op.add_option("-r", "--receivers", default=1, type="int", metavar="N", + help="start N receivers per queue (default %default)") +op.add_option("-m", "--messages", default=100000, type="int", metavar="N", + help="send N messages per sender (default %default)") +op.add_option("--queue-name", default="benchmark", metavar="NAME", + help="base name for queues (default %default)") +op.add_option("--send-rate", default=0, metavar="N", + help="send rate limited to N messages/second, 0 means no limit (default %default)") +op.add_option("--receive-rate", default=0, metavar="N", + help="receive rate limited to N messages/second, 0 means no limit (default %default)") +op.add_option("--content-size", default=1024, type="int", metavar="BYTES", + help="message size in bytes (default %default)") +op.add_option("--ack-frequency", default=100, metavar="N", type="int", + help="receiver ack's every N messages, 0 means unconfirmed (default %default)") +op.add_option("--no-report-header", dest="report_header", default=True, + action="store_false", help="don't print header on report") +op.add_option("--summarize", default=False, action="store_true", + help="print summary statistics for multiple senders/receivers: total throughput, average latency") +op.add_option("--repeat", default=1, metavar="N", help="repeat N times", type="int") +op.add_option("--send-option", default=[], action="append", type="str", + help="Additional option for sending addresses") +op.add_option("--receive-option", default=[], action="append", type="str", + help="Additional option for receiving addresses") +op.add_option("--create-option", default=[], action="append", type="str", + help="Additional option for creating addresses") +op.add_option("--send-arg", default=[], action="append", type="str", + help="Additional argument for qpid-send") +op.add_option("--receive-arg", default=[], action="append", type="str", + help="Additional argument for qpid-receive") +op.add_option("--no-timestamp", dest="timestamp", default=True, + action="store_false", help="don't add a timestamp, no latency results") +op.add_option("--sequence", dest="sequence", default=False, + action="store_true", help="add a sequence number to each message") +op.add_option("--connection-options", type="str", + help="Connection options for senders & receivers") +op.add_option("--durable", default=False, action="store_true", + help="Use durable queues and messages") +op.add_option("--save-received", default=False, action="store_true", + help="Save received message content to files <queuename>-receiver-<n>.msg") +op.add_option("--verbose", default=False, action="store_true", + help="Show commands executed") +op.add_option("--no-delete", default=False, action="store_true", + help="Don't delete the test queues.") +op.add_option("--fill-drain", default=False, action="store_true", + help="First fill the queues, then drain them") + +single_quote_re = re.compile("'") +def posix_quote(string): + """ Quote a string for use as an argument in a posix shell""" + return "'" + single_quote_re.sub("\\'", string) + "'"; + +def ssh_command(host, command): + """ Convert command into an ssh command on host with quoting""" + return ["ssh", host] + [posix_quote(arg) for arg in command] + +class Clients: + def __init__(self): self.clients=[] + + def add(self, client): + self.clients.append(client) + return client + + def kill(self): + for c in self.clients: + try: c.kill() + except: pass + +class PopenCommand(Popen): + """Like Popen but you can query for the command""" + def __init__(self, command, *args, **kwargs): + self.command = command + Popen.__init__(self, command, *args, **kwargs) + +clients = Clients() + +def start_receive(queue, index, opts, ready_queue, broker, host): + address_opts=opts.receive_option + if opts.durable: address_opts += ["node:{durable:true}"] + address="%s;{%s}"%(queue,",".join(address_opts)) + msg_total=opts.senders*opts.messages + messages = msg_total/opts.receivers; + if (index < msg_total%opts.receivers): messages += 1 + if (messages == 0): return None + command = ["qpid-jms-receive", + #"-b", broker, + "--ready-address", "benchmark-ready;{create:always}", + "-a", address, + "-m", str(messages), + "--forever", + "--print-content=no", + # "--receive-rate", str(opts.receive_rate), + "--report-total", + "--ack-frequency", str(opts.ack_frequency), + # "--ready-address", "%s;{create:always}"%ready_queue, + "--report-header=no -v" + ] + if opts.save_received: + command += ["--save-content=%s-receiver-%s.msg"%(queue,index)] + command += opts.receive_arg + if opts.connection_options: + command += ["--connection-options",opts.connection_options] + if host: command = ssh_command(host, command) + if opts.verbose: print "Receiver: ", command + return clients.add(PopenCommand(command, stdout=PIPE, stderr=PIPE)) + +def start_send(queue, opts, broker, host): + address="%s;{%s}"%(queue,",".join(opts.send_option + ["create:always"])) + command = ["qpid-jms-send", + #"-b", broker, + "-a", address, + "--messages", str(opts.messages), + "--content-size", str(opts.content_size), + "--send-rate", str(opts.send_rate), + "--report-total", + "--report-header=no", + "--timestamp=%s"%(opts.timestamp and "yes" or "no"), + "--sequence=%s"%(opts.sequence and "yes" or "no"), + "--durable", str(opts.durable) + ] + command += opts.send_arg + if opts.connection_options: + command += ["--connection-options",opts.connection_options] + if host: command = ssh_command(host, command) + if opts.verbose: print "Sender: ", command + return clients.add(PopenCommand(command, stdout=PIPE, stderr=PIPE)) + +def error_msg(out, err): + return ("\n[stdout]\n%s\n[stderr]\n%s[end]"%(out, err)) + +def first_line(p): + out,err=p.communicate() + if p.returncode != 0: + raise Exception("Process exit %d: %s"%(p.returncode, error_msg(out,err))) + + print str(out) + print str(err) + return out.split("\n")[0] + +def recreate_queues(queues, brokers, no_delete, opts): + c = qpid.messaging.Connection(brokers[0]) + c.open() + s = c.session() + for q in queues: + if not no_delete: + try: s.sender("%s;{delete:always}"%(q)).close() + except qpid.messaging.exceptions.NotFound: pass + address = "%s;{%s}"%(q, ",".join(opts.create_option + ["create:always"])) + if opts.verbose: print "Creating", address + s.sender(address) + c.close() + +def print_header(timestamp): + if timestamp: latency_header="\tl-min\tl-max\tl-avg\ttotal-tp" + else: latency_header="" + print "send-tp\trecv-tp%s"%latency_header + +def parse(parser, lines): # Parse sender/receiver output + return [map(lambda p: p[0](p[1]), zip(parser,line.split())) for line in lines] + +def parse_senders(senders): + return parse([int],[first_line(p) for p in senders]) + +def parse_receivers(receivers): + return parse([int,float,float,float],[first_line(p) for p in receivers if p]) + +def print_data(send_stats, recv_stats, total_tp): + for send,recv in map(None, send_stats, recv_stats): + line="" + if send: line += "%d"%send[0] + if recv: + line += "\t%d"%recv[0] + if len(recv) == 4: line += "\t%.2f\t%.2f\t%.2f"%tuple(recv[1:]) + if total_tp is not None: + line += "\t%d"%total_tp + total_tp = None + print line + +def print_summary(send_stats, recv_stats, total_tp): + def avg(s): sum(s) / len(s) + send_tp = sum([l[0] for l in send_stats]) + recv_tp = sum([l[0] for l in recv_stats]) + summary = "%d\t%d"%(send_tp, recv_tp) + if recv_stats and len(recv_stats[0]) == 4: + l_min = sum(l[1] for l in recv_stats)/len(recv_stats) + l_max = sum(l[2] for l in recv_stats)/len(recv_stats) + l_avg = sum(l[3] for l in recv_stats)/len(recv_stats) + summary += "\t%.2f\t%.2f\t%.2f"%(l_min, l_max, l_avg) + summary += "\t%d"%total_tp + print summary + + +class ReadyReceiver: + """A receiver for ready messages""" + def __init__(self, queue, broker): + self.connection = qpid.messaging.Connection(broker) + self.connection.open() + self.receiver = self.connection.session().receiver( + "%s;{create:receiver,delete:receiver,node:{durable:false}}"%(queue)) + self.receiver.session.sync() + self.timeout=10 + + def wait(self, receivers): + try: + for i in receivers: self.receiver.fetch(self.timeout) + self.connection.close() + except qpid.messaging.Empty: + for r in receivers: + if (r.poll() is not None): + out,err=r.communicate() + raise Exception("Receiver error: %s\n%s" % + (" ".join(r.command), error_msg(out,err))) + raise Exception("Timed out waiting for receivers to be ready") + +def flatten(l): + return sum(map(lambda s: re.split(re.compile("\s*,\s*|\s+"), s), l), []) + +class RoundRobin: + def __init__(self,items): + self.items = items + self.index = 0 + + def next(self): + if not self.items: return None + ret = self.items[self.index] + self.index = (self.index+1)%len(self.items) + return ret + +def main(): + opts, args = op.parse_args() + opts.client_host = flatten(opts.client_host) + if not opts.broker: + if opts.client_host: + raise Exception("--broker must be specified if --client_host is.") + opts.broker = ["127.0.0.1"] # Deafult to local broker + opts.broker = flatten(opts.broker) + brokers = RoundRobin(opts.broker) + client_hosts = RoundRobin(opts.client_host) + send_out = "" + receive_out = "" + ready_queue="%s-ready"%(opts.queue_name) + queues = ["%s-%s"%(opts.queue_name, i) for i in xrange(opts.queues)] + try: + for i in xrange(opts.repeat): + recreate_queues(queues, opts.broker, opts.no_delete, opts) + ready_receiver = ReadyReceiver(ready_queue, opts.broker[0]) + + def start_receivers(): + return [ start_receive(q, j, opts, ready_queue, brokers.next(), client_hosts.next()) + for q in queues for j in xrange(opts.receivers) ] + + + def start_senders(): + return [ start_send(q, opts,brokers.next(), client_hosts.next()) + for q in queues for j in xrange(opts.senders) ] + + if opts.report_header and i == 0: print_header(opts.timestamp) + + if opts.fill_drain: + # First fill the queues, then drain them + start = time.time() + senders = start_senders() + for p in senders: p.wait() + receivers = start_receivers() + for p in receivers: p.wait() + else: + # Run senders and receivers in parallel + receivers = start_receivers() + ready_receiver.wait(filter(None, receivers)) # Wait for receivers ready + start = time.time() + senders = start_senders() + for p in senders + receivers: p.wait() + + total_sent = opts.queues * opts.senders * opts.messages + total_tp = total_sent / (time.time()-start) + #send_stats=parse_senders(senders) + recv_stats=parse_receivers(receivers) + #if opts.summarize: print_summary(send_stats, recv_stats, total_tp) + #else: print_data(send_stats, recv_stats, total_tp) + finally: clients.kill() # No strays + +if __name__ == "__main__": main() + diff --git a/java/tools/bin/qpid-jms-receive b/java/tools/bin/qpid-jms-receive new file mode 100755 index 0000000000..57abe874ff --- /dev/null +++ b/java/tools/bin/qpid-jms-receive @@ -0,0 +1,193 @@ +#!/bin/sh +# +# 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. +# + +# This starts the controller for coordinating perf tests/ + +. check-qpid-java-env + +PROGRAM_NAME="qpid-jms-receive" +URL="amqp://guest:guest@clientid/testpath?brokerlist='tcp://localhost:5672'" +ADDRESS="queue;{create:always}" +TIMEOUT="0" +FOREVER="false" +MESSAGES="1" +IGNORE_DUPLICATES="false" +CHECK_REDELIVERED="false" +CAPACITY="1000" +ACK_FREQUENCY="100" +TX="0" +ROLLBACL_FREQUENCY="0" +PRINT_CONTENT="false" +PRINT_HEADERS="false" +REPORT_TOTAL="false" +REPORT_EVERY="0" +REPORT_HEADER="true" +READY_ADDRES="''" +EXTRA_JVM_ARGS="" +VERBOSE="0" + +TEST_ID=`echo ${HOSTNAME} | awk -F . '{print $1}'` + +TEMP=$(getopt -n $PROGRAM_NAME -o b:a:f:m:vh\ + --long broker:,address:,timeout:,forever\ +,messages:,ignore-duplicates,check-redelivered\ +,capacity:,ack-frequency:,tx:,rollback-frequency:\ +,print-content:,print-headers:,report-total\ +,report-every:,report-header:,ready-address:\ +,jvm-args:,verbose,help -- "$@") + +# padding the option string with 4 spaces +# padding the desc string with 30 spaces +usage() +{ + printf "\n%s\n" "Usage: $PROGRAM_NAME [option].." + + printf "\n%20s\n%57s\n" "-b, --broker URL" "url of broker to connect to" + + printf "\n%24s\n%53s\n" "-a,--address ADDRESS" "address to receive from" + + printf "\n%25s\n%71s\n" "--timeout TIMEOUT (0)" "timeout in seconds to wait before exiting" + + printf "\n%17s\n%61s\n" "-f, --forever" "ignore timeout and wait forever" + + printf "\n%24s\n%89s\n" "-m, --messages N (0)" "Number of messages to receive; 0 means receive indefinitely" + + printf "\n%23s\n%84s\n" "--ignore-duplicates" "Detect and ignore duplicates (by checking 'sn' header)" + + printf "\n%23s\n%82s\n%92s\n" "--check-redelivered" "Fails with exception if a duplicate is not marked as" " redelivered (only relevant when ignore-duplicates is selected)" + + printf "\n%23s\n%71s\n" "--capacity N (1000)" "Pre-fetch window (0 implies no pre-fetch)" + + printf "\n%27s\n%94s\n" "--ack-frequency N (100)" "Ack frequency (0 implies none of the messages will get accepted)" + + printf "\n%14s\n%94s\n" "--tx N (0)" "batch size for transactions (0 implies transaction are not used)" + + printf "\n%30s\n%94s\n" "--rollback-frequency N (0)" "rollback frequency (0 implies no transaction will be rolledback)" + + printf "\n%30s\n%55s\n" "--print-content yes|no (0)" "print out message content" + + printf "\n%30s\n%55s\n" "--print-headers yes|no (0)" "print out message headers" + + printf "\n%18s\n%76s\n" "--report-total" "Report total throughput and latency statistics" + + printf "\n%24s\n%87s\n" "--report-every N (0)" "Report throughput and latency statistics every N messages" + + printf "\n%30s\n%47s\n" "--report-header yes|no (1)" "Headers on report" + + printf "\n%27s\n%82s\n" "--ready-address ADDRESS" "send a message to this address when ready to receive" + + printf "\n%14s\n%69s\n" "--jvm-args" "Extra jvm arguments you want to specify" + + printf "\n%17s\n%69s\n\n" "-v, --verbose" "Print debug information for this script" +} + +eval set -- "$TEMP" +while true; do + case $1 in + -b|--broker) + URL="$2"; shift; shift; continue + ;; + -a|--address) + ADDRESS="$2"; shift; shift; continue + ;; + --timeout) + TIMEOUT="$2"; shift; shift; continue + ;; + -f|--forever) + FOREVER="$2"; shift; shift; continue + ;; + -m|--messages) + MESSAGES="$2"; shift; shift; continue + ;; + --ignore-duplicates) + IGNORE_DUPLICATES="true"; shift; continue + ;; + --check-redelivered) + CHECK_REDELIVERED="true"; shift; continue + ;; + --capacity) + CAPACITY="$2"; shift; shift; continue + ;; + --ack-frequency) + ACK_FREQUENCY="$2"; shift; shift; continue + ;; + --tx) + TX="$2"; shift; shift; continue + ;; + --rollback-frequency) + ROLLBACK_FREQUENCY="$2"; shift; shift; continue + ;; + --print-content) + if [ "$2" == "yes" ]; then PRINT_CONTENT="true"; else PRINT_CONTENT="false"; fi; shift; shift; continue + ;; + --print-headers) + if [ "$2" == "yes" ]; then PRINT_HEADERS="true"; else PRINT_HEADERS="false"; fi; shift; shift; continue + ;; + --report-total) + REPORT_TOTAL="true"; shift; continue + ;; + --report-every) + REPORT_EVERY="$2"; shift; shift; continue + ;; + --report-header) + if [ "$2" == "yes" ]; then REPORT_HEADER="true"; else REPORT_HEADER="false"; fi; shift; shift; continue + ;; + --ready-address) + READY_ADDRESS="$2"; shift; shift; continue + ;; + -a|--jvm-args) + EXTRA_JVM_ARGS="$2"; shift; shift; continue + ;; + -h|--help) + usage + exit 0 + ;; + -v|--verbose) + VERBOSE="1"; shift; continue + ;; + --) + # no more arguments to parse + break + ;; + *) + # no more arguments to parse + break + ;; + esac +done + +RECEIVER_ARGS="-server -Durl=$URL \ +-Daddress=$ADDRESS \ +-Dtimeout=$TIMEOUT \ +-Dmsg-count=$MESSAGES \ +-Dack-frequency=$ACK_FREQUENCY \ +-Dtx=$TX \ +-Drollback-frequnecy=$ROLLBACL_FREQUENCY \ +-Dprint-content=$PRINT_CONTENT \ +-Dprint-headers=$PRINT_HEADERS \ +-Dreport-total=$REPORT_TOTAL \ +-Dreport-every=$REPORT_EVERY \ +-Dreport-header=$REPORT_HEADER \ +-Dmax_prefetch=$CAPACITY " + +if [ "x$READY_ADDRESS" != "x" ]; then RECEIVER_ARGS="$RECEIVER_ARGS -Dready-address=$READY_ADDRESS"; fi +if [ "$VERBOSE" == "1" ]; then echo $RECEIVER_ARGS; fi +echo $RECEIVER_ARGS +$JAVA -cp $CLASSPATH $LOG_CONFIG $JAVA_MEM $RECEIVER_ARGS org.apache.qpid.tools.QpidReceive diff --git a/java/tools/bin/qpid-jms-send b/java/tools/bin/qpid-jms-send new file mode 100755 index 0000000000..d7695924f0 --- /dev/null +++ b/java/tools/bin/qpid-jms-send @@ -0,0 +1,261 @@ +#!/bin/sh +# +# 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. +# + +# This starts the controller for coordinating perf tests/ + +. check-qpid-java-env + +PROGRAM_NAME="qpid-jms-send" +URL="amqp://guest:guest@clientid/testpath?brokerlist='tcp://localhost:5672'" +ADDRESS="queue;{create:always}" +MESSAGES="1" +ID="" +REPLY_TO="" +SEND_EOS="1" +DURABLE="false" +TTL="0" +PRIORITY="0" +PROPERTY="" +CORRELATION_ID="" +USER_ID="" +CONTENT_STRING="" +CONTENT_SIZE="1024" +CONTENT_MAP="" +CAPACITY="1000" +ACK_FREQUENCY="100" +TX="0" +ROLLBACL_FREQUENCY="0" +PRINT_CONTENT="true" +PRINT_HEADERS="false" +REPORT_TOTAL="false" +REPORT_EVERY="0" +REPORT_HEADER="true" +SEND_RATE="-1" +SEQUNCE="1" +DISABLE_TIMESTAMP="false" +EXTRA_JVM_ARGS="" +VERBOSE="0" + +TEST_ID=`echo ${HOSTNAME} | awk -F . '{print $1}'` + +TEMP=$(getopt -n $PROGRAM_NAME -o b:a:m:i:P:M:vh\ + --long broker:,address:,messages:,id:,reply-to:\ +,send-eos:,durable:,ttl:,property:,correlational-id:\ +,user-id:,content-string:,content-size:,content-map:\ +,capacity:,ack-frequency:,tx:,rollback-frequency:\ +,print-content:,print-headers:,report-total\ +,report-every:,report-header:,send-rate:,sequence:,timestamp:\ +,jvm-args:,verbose,help -- "$@") + +# padding the option string with 4 spaces +# padding the desc string with 30 spaces +usage() +{ + printf "\n%s\n" "Usage: $PROGRAM_NAME [option].." + + printf "\n%20s\n%57s\n" "-b, --broker URL" "url of broker to connect to" + + printf "\n%24s\n%53s\n" "-a,--address ADDRESS" "address to receive from" + + printf "\n%24s\n%89s\n" "-m, --messages N (0)" "Number of messages to receive; 0 means receive indefinitely" + + printf "\n%15s\n%75s\n" "-i, --id ID" "use the supplied id instead of generating one" + + printf "\n%23s\n%54s\n" "--reply-to REPLY-TO" "specify reply-to address" + + printf "\n%20s\n%70s\n" "--send-eos N (0)" "send N EOS messages to mark end of input" + + printf "\n%24s\n%54s\n" "--durable yes|no (0)" "mark messages as durable" + + printf "\n%19s\n%72s\n" "--ttl msecs (0)" "time-to-live for messages, in milliseconds" + + printf "\n%27s\n%72s\n" "--priority PRIORITY (0)" "time-to-live for messages, in milliseconds" + + printf "\n%29s\n%54s\n" "-P, --property NAME=VALUE" "specify message property" + + printf "\n%23s\n%57s\n" "--correlation-id ID" "correlation-id for message" + + printf "\n%20s\n%48s\n" "--user-id USERID" "userid for message" + + printf "\n%28s\n%60s\n" "--content-string CONTENT" "use CONTENT as message content" + + printf "\n%24s\n%62s\n" "--content-size N (0)" "create an N-byte message content" + + printf "\n%32s\n%59s\n" "-M, --content-map NAME=VALUE" "specify entry for map content" + + printf "\n%23s\n%71s\n" "--capacity N (1000)" "Pre-fetch window (0 implies no pre-fetch)" + + printf "\n%27s\n%94s\n" "--ack-frequency N (100)" "Ack frequency (0 implies none of the messages will get accepted)" + + printf "\n%14s\n%94s\n" "--tx N (0)" "batch size for transactions (0 implies transaction are not used)" + + printf "\n%30s\n%94s\n" "--rollback-frequency N (0)" "rollback frequency (0 implies no transaction will be rolledback)" + + printf "\n%30s\n%55s\n" "--print-content yes|no (1)" "print out message content" + + printf "\n%30s\n%55s\n" "--print-headers yes|no (0)" "print out message headers" + + printf "\n%18s\n%76s\n" "--report-total" "Report total throughput and latency statistics" + + printf "\n%24s\n%87s\n" "--report-every N (0)" "Report throughput and latency statistics every N messages" + + printf "\n%30s\n%47s\n" "--report-header yes|no (1)" "Headers on report" + + printf "\n%21s\n%64s\n%62s\n" "--send-rate N (0)" "Send at rate of N messages/second." "0 means send as fast as possible" + + printf "\n%25s\n%69s\n%77s\n" "--sequence yes|no (1)" "Add a sequence number messages property" "(required for duplicate/lost message detection)" + + printf "\n%26s\n%64s\n%77s\n" "--timestamp yes|no (1)" "Add a time stamp messages property" "(required for duplicate/lost message detection)" + + printf "\n%14s\n%69s\n" "--jvm-args" "Extra jvm arguments you want to specify" + + printf "\n%17s\n%69s\n\n" "-v, --verbose" "Print debug information for this script" +} + +eval set -- "$TEMP" +while true; do + case $1 in + -b|--broker) + URL="$2"; shift; shift; continue + ;; + -a|--address) + ADDRESS="$2"; shift; shift; continue + ;; + -m|--messages) + MESSAGES="$2"; shift; shift; continue + ;; + -i|--id) + ID="$2"; shift; shift; continue + ;; + --reply-to) + REPLY_TO="$2"; shift; shift; continue + ;; + --send-eos) + SEND_EOS="$2"; shift; shift; continue + ;; + --durable) + if [ "$2" == "1" ]; then DURABLE="true"; else DURABLE="false"; fi; shift; shift; continue + ;; + --ttl) + TTL="$2"; shift; shift; continue + ;; + --priority) + PRIORITY="$2"; shift; shift; continue + ;; + -P|--property) + PROPERTY="$2,$PROPERTY"; shift; shift; continue + ;; + --correlation-id) + CORRELATION_ID="$2"; shift; shift; continue + ;; + --user-id) + USER_ID="$2"; shift; shift; continue + ;; + --content-string) + CONTENT_STRING="$2"; shift; shift; continue + ;; + --content-size) + CONTENT_SIZE="$2"; shift; shift; continue + ;; + -M|--content-map) + CONTENT_MAP="$2,$CONTENT_MAP"; shift; shift; continue + ;; + --capacity) + CAPACITY="$2"; shift; shift; continue + ;; + --ack-frequency) + ACK_FREQUENCY="$2"; shift; shift; continue + ;; + --tx) + TX="$2"; shift; shift; continue + ;; + --rollback-frequency) + ROLLBACK_FREQUENCY="$2"; shift; shift; continue + ;; + --print-content) + if [ "$2" == "yes" ]; then PRINT_CONTENT="true"; else PRINT_CONTENT="false"; fi; shift; shift; continue + ;; + --print-headers) + if [ "$2" == "yes" ]; then PRINT_HEADERS="true"; else PRINT_HEADERS="false"; fi; shift; shift; continue + ;; + --report-total) + REPORT_TOTAL="true"; shift; continue + ;; + --report-every) + REPORT_EVERY="$2"; shift; shift; continue + ;; + --report-header) + if [ "$2" == "yes" ]; then REPORT_HEADER="true"; else REPORT_HEADER="false"; fi; shift; shift; continue + ;; + --send-rate) + SEND_RATE="$2"; shift; shift; continue + ;; + --sequence) + if [ "$2" == "yes" ]; then SEQUENCE="true"; else SEQUENCE="false"; fi; shift; shift; continue + ;; + --timestamp) + if [ "$2" == "yes" ]; then DISABLE_TIMESTAMP="false"; else DISABLE_TIMESTAMP="true"; fi; shift; shift; continue + ;; + -a|--jvm-args) + EXTRA_JVM_ARGS="$2"; shift; shift; continue + ;; + -h|--help) + usage + exit 0 + ;; + -v|--verbose) + VERBOSE="1"; shift; continue + ;; + --) + # no more arguments to parse + break + ;; + *) + # no more arguments to parse + break + ;; + esac +done + +SENDER_ARGS="-server -Durl=$URL \ +-Daddress=$ADDRESS \ +-Dmsg-count=$MESSAGES \ +-Dsend-eos=$SEND_EOS \ +-Ddurable=$DURABLE \ +-Dmsg_size=$CONTENT_SIZE \ +-Dsend-rate=$SEND_RATE \ +-Ddisable-timestamp=$DISABLE_TIMESTAMP \ +-Dttl=$TTL \ +-Dpriority=$PRIORITY \ +-Dtx=$TX \ +-Drollback-frequnecy=$ROLLBACK_FREQUENCY \ +-Dprint-content=$PRINT_CONTENT \ +-Dprint-headers=$PRINT_HEADERS \ +-Dreport-total=$REPORT_TOTAL \ +-Dreport-every=$REPORT_EVERY \ +-Dreport-header=$REPORT_HEADER \ +-Dmax_prefetch=$CAPACITY " + +if [ "x$ID" != "x" ]; then SENDER_ARGS="$SENDER_ARGS -Did=$ID"; fi +if [ "x$USER_ID" != "x" ]; then SENDER_ARGS="$SENDER_ARGS -Duser_id=$USER_ID"; fi +if [ "x$CORRELATION_ID" != "x" ]; then SENDER_ARGS="$SENDER_ARGS -Dcorrelation_id=$CORRELATION_ID"; fi + +if [ "$VERBOSE" == "1" ]; then echo $SENDER_ARGS; fi +$JAVA -cp $CLASSPATH $LOG_CONFIG $JAVA_MEM $SENDER_ARGS org.apache.qpid.tools.QpidSend |