| #!/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, random, os, time, uuid |
| from qpid.messaging import * |
| import statistics |
| |
| EOS = "eos" |
| SN = "sn" |
| TS = "ts" |
| |
| TIME_SEC = 1000000000 |
| SECOND = 1000 |
| |
| def nameval(st): |
| idx = st.find("=") |
| if idx >= 0: |
| name = st[0:idx] |
| value = st[idx+1:] |
| else: |
| name = st |
| value = None |
| return name, value |
| |
| |
| op = optparse.OptionParser(usage="usage: %prog [options]", description="Spouts messages to the specified address") |
| op.add_option("-b", "--broker", default="localhost:5672", type="str", help="url of broker to connect to") |
| op.add_option("-a", "--address", type="str", help="address to send to") |
| op.add_option("--connection-options", default={}, help="options for the connection") |
| op.add_option("-m", "--messages", default=1, type="int", help="stop after N messages have been sent, 0 means no limit") |
| op.add_option("-i", "--id", type="str", help="use the supplied id instead of generating one") |
| op.add_option("--reply-to", type="str", help="specify reply-to address") |
| op.add_option("--send-eos", default=0, type="int", help="Send N EOS messages to mark end of input") |
| op.add_option("--durable", default=False, action="store_true", help="Mark messages as durable") |
| op.add_option("--ttl", default=0, type="int", help="Time-to-live for messages, in milliseconds") |
| op.add_option("--priority", default=0, type="int", help="Priority for messages (higher value implies higher priority)") |
| op.add_option("-P", "--property", default=[], action="append", type="str", help="specify message property") |
| op.add_option("--correlation-id", type="str", help="correlation-id for message") |
| op.add_option("--user-id", type="str", help="userid for message") |
| op.add_option("--content-string", type="str", help="use CONTENT as message content") |
| op.add_option("--content-size", default=0, type="int", help="create an N-byte message content") |
| op.add_option("-M", "--content-map", default=[], action="append", type="str", help="specify entry for map content") |
| op.add_option("--content-stdin", default=False, action="store_true", help="read message content from stdin, one line per message") |
| op.add_option("--capacity", default=1000, type="int", help="size of the senders outgoing message queue") |
| op.add_option("--tx", default=0, type="int", help="batch size for transactions (0 implies transaction are not used)") |
| op.add_option("--rollback-frequency", default=0, type="int", help="rollback frequency (0 implies no transaction will be rolledback)") |
| op.add_option("--failover-updates", default=False, action="store_true", help="Listen for membership updates distributed via amq.failover") |
| op.add_option("--report-total", default=False, action="store_true", help="Report total throughput statistics") |
| op.add_option("--report-every", default=0, type="int", help="Report throughput statistics every N messages") |
| op.add_option("--report-header", type="str", default="yes", help="Headers on report") |
| op.add_option("--send-rate", default=0, type="int", help="Send at rate of N messages/second. 0 means send as fast as possible") |
| op.add_option("--flow-control", default=0, type="int", help="Do end to end flow control to limit queue depth to 2*N. 0 means no flow control.") |
| op.add_option("--sequence", type="str", default="yes", help="Add a sequence number messages property (required for duplicate/lost message detection)") |
| op.add_option("--timestamp", type="str", default="yes", help="Add a time stamp messages property (required for latency measurement)") |
| op.add_option("--group-key", type="str", help="Generate groups of messages using message header 'KEY' to hold the group identifier") |
| op.add_option("--group-prefix", default="GROUP-", type="str", help="Generate group identifers with 'STRING' prefix (if group-key specified)") |
| op.add_option("--group-size", default=10, type="int", help="Number of messages per a group (if group-key specified)") |
| op.add_option("--group-randomize-size", default=False, action="store_true", help="Randomize the number of messages per group to [1...group-size] (if group-key specified)") |
| op.add_option("--group-interleave", default=1, type="int", help="Simultaineously interleave messages from N different groups (if group-key specified)") |
| |
| |
| class ContentGenerator: |
| def setContent(self, msg): |
| return |
| |
| class GetlineContentGenerator(ContentGenerator): |
| def setContent(self, msg): |
| content = sys.stdin.readline() |
| got = (not line) |
| if (got): |
| msg.content = content |
| return got |
| |
| class FixedContentGenerator(ContentGenerator): |
| def __init__(self, content=None): |
| self.content = content |
| |
| def setContent(self, msg): |
| msg.content = self.content |
| return True |
| |
| class MapContentGenerator(ContentGenerator): |
| def __init__(self, opts=None): |
| self.opts = opts |
| |
| def setContent(self, msg): |
| self.content = {} |
| for e in self.opts.content_map: |
| name, val = nameval(p) |
| content[name] = val |
| msg.content = self.content |
| return True |
| |
| |
| # tag each generated message with a group identifer |
| class GroupGenerator: |
| def __init__(self, key, prefix, size, randomize, interleave): |
| groupKey = key |
| groupPrefix = prefix |
| groupSize = size |
| randomizeSize = randomize |
| groupSuffix = 0 |
| if (randomize > 0): |
| random.seed(os.getpid()) |
| |
| for i in range(0, interleave): |
| self.newGroup() |
| current = 0 |
| |
| def setGroupInfo(self, msg): |
| if (current == len(groups)): |
| current = 0 |
| my_group = groups[current] |
| msg.properties[groupKey] = my_group[id]; |
| # print "SENDING GROUPID=[%s]\n" % my_group[id] |
| my_group[count]=my_group[count]+1 |
| if (my_group[count] == my_group[size]): |
| self.newGroup() |
| del groups[current] |
| else: |
| current+=1 |
| |
| def newGroup(self): |
| groupId = "%s%s" % (groupPrefix, groupSuffix) |
| groupSuffix+=1 |
| size = groupSize |
| if (randomizeSize == True): |
| size = random.randint(1,groupSize) |
| # print "New group: GROUPID=["%s] size=%s" % (groupId, size) |
| groups.append({'id':groupId, 'size':size, 'count':0}) |
| |
| |
| |
| def main(): |
| opts, args = op.parse_args() |
| if not opts.address: |
| raise Exception("Address must be specified!") |
| |
| broker = opts.broker |
| address = opts.address |
| connection = Connection(opts.broker, **opts.connection_options) |
| |
| try: |
| connection.open() |
| if (opts.failover_updates): |
| auto_fetch_reconnect_urls(connection) |
| session = connection.session(transactional=(opts.tx)) |
| sender = session.sender(opts.address) |
| if (opts.capacity>0): |
| sender.capacity = opts.capacity |
| sent = 0 |
| txCount = 0 |
| stats = statistics.Throughput() |
| reporter = statistics.Reporter(opts.report_every, opts.report_header == "yes", stats) |
| |
| contentGen = ContentGenerator() |
| content = "" # auxiliary variable for determining content type of message - needs to be changed to {} for Map message |
| if opts.content_stdin: |
| opts.messages = 0 # Don't limit number of messages sent. |
| contentGen = GetlineContentGenerator() |
| elif opts.content_map is not None: |
| contentGen = MapContentGenerator(opts) |
| content = {} |
| elif opts.content_size is not None: |
| contentGen = FixedContentGenerator('X' * opts.content_size) |
| else: |
| contentGen = FixedContentGenerator(opts.content_string) |
| if opts.group_key is not None: |
| groupGen = GroupGenerator(opts.group_key, opts.group_prefix, opts.group_size, opts.group_random_size, opts.group_interleave) |
| |
| msg = Message(content=content) |
| msg.durable = opts.durable |
| if opts.ttl: |
| msg.ttl = opts.ttl/1000.0 |
| if opts.priority: |
| msg.priority = opts.priority |
| if opts.reply_to is not None: |
| if opts.flow_control > 0: |
| raise Exception("Can't use reply-to and flow-control together") |
| msg.reply_to = opts.reply_to |
| if opts.user_id is not None: |
| msg.user_id = opts.user_id |
| if opts.correlation_id is not None: |
| msg.correlation_id = opts.correlation_id |
| for p in opts.property: |
| name, val = nameval(p) |
| msg.properties[name] = val |
| |
| start = time.time()*TIME_SEC |
| interval = 0 |
| if opts.send_rate > 0: |
| interval = TIME_SEC/opts.send_rate |
| |
| flowControlAddress = "flow-" + str(uuid.uuid1()) + ";{create:always,delete:always}" |
| flowSent = 0 |
| if opts.flow_control > 0: |
| flowControlReceiver = session.receiver(flowControlAddress) |
| flowControlReceiver.capacity = 2 |
| |
| while (contentGen.setContent(msg) == True): |
| sent+=1 |
| if opts.sequence == "yes": |
| msg.properties[SN] = sent |
| |
| if opts.flow_control > 0: |
| if (sent % opts.flow_control == 0): |
| msg.reply_to = flowControlAddress |
| flowSent+=1 |
| else: |
| msg.reply_to = "" # Clear the reply address. |
| |
| if 'groupGen' in vars(): |
| groupGen.setGroupInfo(msg) |
| |
| if (opts.timestamp == "yes"): |
| msg.properties[TS] = int(time.time()*TIME_SEC) |
| sender.send(msg) |
| reporter.message(msg) |
| |
| if ((opts.tx > 0) and (sent % opts.tx == 0)): |
| txCount+=1 |
| if ((opts.rollbackFrequency > 0) and (txCount % opts.rollbackFrequency == 0)): |
| session.rollback() |
| else: |
| session.commit() |
| if ((opts.messages > 0) and (sent >= opts.messages)): |
| break |
| |
| if (opts.flow_control > 0) and (flowSent == 2): |
| flowControlReceiver.fetch(timeout=SECOND) |
| flowSent -= 1 |
| |
| if (opts.send_rate > 0): |
| delay = start + sent*interval - time.time()*TIME_SEC |
| if (delay > 0): |
| time.sleep(delay) |
| #end of while |
| |
| while flowSent > 0: |
| flowControlReceiver.fetch(timeout=SECOND) |
| flowSent -= 1 |
| |
| if (opts.report_total): |
| reporter.report() |
| for i in reversed(range(1,opts.send_eos+1)): |
| if (opts.sequence == "yes"): |
| sent+=1 |
| msg.properties[SN] = sent |
| msg.properties[EOS] = True #TODO (also in C++ client): add in ability to send digest or similar |
| sender.send(msg) |
| if ((opts.tx > 0) and (sent % opts.tx == 0)): |
| txCount+=1 |
| if ((opts.rollback_frequency > 0) and (txCount % opts.rollback_frequency == 0)): |
| session.rollback() |
| else: |
| session.commit() |
| session.sync() |
| session.close() |
| connection.close() |
| except Exception,e: |
| print e |
| connection.close() |
| |
| if __name__ == "__main__": main() |
