From 248f1fe188fe2307b9dcf2c87a83b653eaa1920c Mon Sep 17 00:00:00 2001 From: "Rafael H. Schloming" Date: Sat, 26 Dec 2009 12:42:57 +0000 Subject: synchronized with trunk except for ruby dir git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid.rnr@893970 13f79535-47bb-0310-9956-ffa450edef68 --- python/examples/api/drain | 62 +++++++++++++++++++++++++++ python/examples/api/server | 87 ++++++++++++++++++++++++++++++++++++++ python/examples/api/spout | 103 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 252 insertions(+) create mode 100755 python/examples/api/drain create mode 100755 python/examples/api/server create mode 100755 python/examples/api/spout (limited to 'python/examples/api') diff --git a/python/examples/api/drain b/python/examples/api/drain new file mode 100755 index 0000000000..485985f16d --- /dev/null +++ b/python/examples/api/drain @@ -0,0 +1,62 @@ +#!/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 +from qpid.messaging import * +from qpid.util import URL + +parser = optparse.OptionParser(usage="usage: %prog [options] ADDRESS ...", + description="Drain messages from the supplied address.") +parser.add_option("-b", "--broker", default="localhost", + help="connect to specified BROKER (default %default)") +parser.add_option("-t", "--timeout", type=float, default=0, + help="timeout in seconds to wait before exiting (default %default)") +parser.add_option("-f", "--forever", action="store_true", + help="ignore timeout and wait forever") + +opts, args = parser.parse_args() + +url = URL(opts.broker) +if args: + addr = args.pop(0) +else: + parser.error("address is required") +if opts.forever: + timeout = None +else: + timeout = opts.timeout + +# XXX: should make URL default the port for us +conn = Connection.open(url.host, url.port or AMQP_PORT, + username=url.user, password=url.password) +ssn = conn.session() +rcv = ssn.receiver(addr) + +while True: + try: + print rcv.fetch(timeout=timeout) + ssn.acknowledge() + except Empty: + break + except ReceiveError, e: + print e + break + +conn.close() diff --git a/python/examples/api/server b/python/examples/api/server new file mode 100755 index 0000000000..adb2dcf792 --- /dev/null +++ b/python/examples/api/server @@ -0,0 +1,87 @@ +#!/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, sys, traceback +from qpid.messaging import * +from qpid.util import URL +from subprocess import Popen, STDOUT, PIPE +from qpid.log import enable, DEBUG, WARN + +parser = optparse.OptionParser(usage="usage: %prog [options] ADDRESS ...", + description="handle requests from the supplied address.") +parser.add_option("-b", "--broker", default="localhost", + help="connect to specified BROKER (default %default)") +parser.add_option("-v", dest="verbose", action="store_true", help="enable logging") + +opts, args = parser.parse_args() + +if opts.verbose: + enable("qpid", DEBUG) +else: + enable("qpid", WARN) + +url = URL(opts.broker) +if args: + addr = args.pop(0) +else: + parser.error("address is required") + +# XXX: should make URL default the port for us +conn = Connection.open(url.host, url.port or AMQP_PORT, + username=url.user, password=url.password) +conn.reconnect = True +ssn = conn.session() +rcv = ssn.receiver(addr) + +def dispatch(msg): + msg_type = msg.properties.get("type") + if msg_type == "shell": + proc = Popen(msg.content, shell=True, stderr=STDOUT, stdin=PIPE, stdout=PIPE) + output, _ = proc.communicate() + result = Message(output) + result.properties["exit"] = proc.returncode + elif msg_type == "eval": + try: + content = eval(msg.content) + except: + content = traceback.format_exc() + result = Message(content) + else: + result = Message("unrecognized message type: %s" % msg_type) + return result + +while True: + try: + msg = rcv.fetch() + response = dispatch(msg) + snd = ssn.sender(msg.reply_to) + try: + snd.send(response) + except SendError, e: + print e + snd.close() + ssn.acknowledge() + except Empty: + break + except ReceiveError, e: + print e + break + +conn.close() diff --git a/python/examples/api/spout b/python/examples/api/spout new file mode 100755 index 0000000000..6a9b2b6e3d --- /dev/null +++ b/python/examples/api/spout @@ -0,0 +1,103 @@ +#!/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 +from qpid.messaging import * +from qpid.util import URL + +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 + +parser = optparse.OptionParser(usage="usage: %prog [options] ADDRESS [ CONTENT ... ]", + description="Send messages to the supplied address.") +parser.add_option("-b", "--broker", default="localhost", + help="connect to specified BROKER (default %default)") +parser.add_option("-c", "--count", type=int, default=1, + help="stop after count messages have been sent, zero disables (default %default)") +parser.add_option("-t", "--timeout", type=float, default=None, + help="exit after the specified time") +parser.add_option("-i", "--id", help="use the supplied id instead of generating one") +parser.add_option("-r", "--reply-to", help="specify reply-to address") +parser.add_option("-P", "--property", dest="properties", action="append", default=[], + help="specify message property") +parser.add_option("-M", "--map", dest="entries", action="append", default=[], + help="specify map entry for message body") + +opts, args = parser.parse_args() + +url = URL(opts.broker) +if opts.id is None: + spout_id = str(uuid4()) +else: + spout_id = opts.id +if args: + addr = args.pop(0) +else: + parser.error("address is required") + +content = None + +if args: + text = " ".join(args) +else: + text = None + +if opts.entries: + content = {} + if text: + content["text"] = text + for e in opts.entries: + name, val = nameval(e) + content[name] = val +else: + content = text + +# XXX: should make URL default the port for us +conn = Connection.open(url.host, url.port or AMQP_PORT, + username=url.user, password=url.password) +ssn = conn.session() +snd = ssn.sender(addr) + +count = 0 +start = time.time() +while (opts.count == 0 or count < opts.count) and \ + (opts.timeout is None or time.time() - start < opts.timeout): + msg = Message(content, reply_to=opts.reply_to) + msg.properties["spout-id"] = "%s:%s" % (spout_id, count) + for p in opts.properties: + name, val = nameval(p) + msg.properties[name] = val + + try: + snd.send(msg) + count += 1 + print msg + except SendError, e: + print e + break + +conn.close() -- cgit v1.2.1