diff options
27 files changed, 641 insertions, 527 deletions
diff --git a/cpp/examples/examples/Makefile b/cpp/examples/examples/Makefile index 8591bd3361..738675c58f 100644 --- a/cpp/examples/examples/Makefile +++ b/cpp/examples/examples/Makefile @@ -1,4 +1,4 @@ -SUBDIRS=direct fanout pub-sub request-response +SUBDIRS=direct fanout pub-sub request-response xml-exchange all: for d in $(SUBDIRS); do ( cd $$d; $(MAKE) $@; ) ; done clean: diff --git a/cpp/examples/examples/direct/listener.cpp b/cpp/examples/examples/direct/listener.cpp index ae6a7699f1..7ee68ebf35 100644 --- a/cpp/examples/examples/direct/listener.cpp +++ b/cpp/examples/examples/direct/listener.cpp @@ -20,7 +20,7 @@ */ /** - * listener.cpp: This program reads messages fro a queue on + * listener.cpp: This program reads messages from a queue on * the broker using a message listener. */ diff --git a/cpp/examples/examples/fanout/listener.cpp b/cpp/examples/examples/fanout/listener.cpp index 79809d679e..c3123bb944 100644 --- a/cpp/examples/examples/fanout/listener.cpp +++ b/cpp/examples/examples/fanout/listener.cpp @@ -69,16 +69,20 @@ int main(int argc, char** argv) { //--------- Main body of program -------------------------------------------- - // Unique name for private queue: + // Each client creates its own private queue, using the + // session id to guarantee a unique name. It then routes + // all messages from the fanout exchange to its own queue + // by binding to the queue. + // + // The binding specifies a binding key, but for a fanout + // exchange, the binding key is optional and is not used + // for routing decisions. It can be useful for tracking + // messages and routing in logs. + std::string myQueue=session.getId().str(); - // Declare my queue. session.queueDeclare(arg::queue=myQueue, arg::exclusive=true, arg::autoDelete=true); - // Bind my queue to the fanout exchange. - //Note no the binding key will not affect routing (its just - //used to identify the binding e.g. when unbinding), the - //fanout exchange delivers all messages to all bound queues - //unconditionally. + session.exchangeBind(arg::exchange="amq.fanout", arg::queue=myQueue, arg::bindingKey="my-key"); // Create a listener and subscribe it to my queue. diff --git a/cpp/examples/examples/fanout/verify_cpp_python.in b/cpp/examples/examples/fanout/verify_cpp_python.in index 6a4c2616d8..21bafe06de 100644 --- a/cpp/examples/examples/fanout/verify_cpp_python.in +++ b/cpp/examples/examples/fanout/verify_cpp_python.in @@ -1,31 +1,27 @@ ==== fanout_producer.out ==== fanout_consumer.py.out | remove_uuid -Messages queue: Subscribed to queue -Response: Message 0 -Response: Message 1 -Response: Message 2 -Response: Message 3 -Response: Message 4 -Response: Message 5 -Response: Message 6 -Response: Message 7 -Response: Message 8 -Response: Message 9 -Response: That's all, folks! -No more messages! +Message 0 +Message 1 +Message 2 +Message 3 +Message 4 +Message 5 +Message 6 +Message 7 +Message 8 +Message 9 +That's all, folks! ==== fanout_consumer.pyX.out | remove_uuid -Messages queue: Subscribed to queue -Response: Message 0 -Response: Message 1 -Response: Message 2 -Response: Message 3 -Response: Message 4 -Response: Message 5 -Response: Message 6 -Response: Message 7 -Response: Message 8 -Response: Message 9 -Response: That's all, folks! -No more messages! +Message 0 +Message 1 +Message 2 +Message 3 +Message 4 +Message 5 +Message 6 +Message 7 +Message 8 +Message 9 +That's all, folks! diff --git a/cpp/examples/examples/pub-sub/topic_listener.cpp b/cpp/examples/examples/pub-sub/topic_listener.cpp index c7e9d3877d..883d6eba42 100644 --- a/cpp/examples/examples/pub-sub/topic_listener.cpp +++ b/cpp/examples/examples/pub-sub/topic_listener.cpp @@ -97,7 +97,7 @@ void Listener::prepareQueue(std::string queue, std::string routing_key) { /* Declare an exclusive queue on the broker */ - session.queueDeclare(arg::queue=queue, arg::exclusive=true); + session.queueDeclare(arg::queue=queue, arg::exclusive=true, arg::autoDelete=true); /* Route messages to the new queue if they match the routing key. * diff --git a/cpp/examples/examples/pub-sub/verify_cpp_python.in b/cpp/examples/examples/pub-sub/verify_cpp_python.in index d16eab0f6c..951d9ad9dd 100644 --- a/cpp/examples/examples/pub-sub/verify_cpp_python.in +++ b/cpp/examples/examples/pub-sub/verify_cpp_python.in @@ -40,11 +40,15 @@ Message 4 Message 4 Message 4 Message 4 -Messages queue: europe -Messages queue: news -Messages queue: usa -Messages queue: weather +Messages on 'europe' queue: +Messages on 'news' queue: +Messages on 'usa' queue: +Messages on 'weather' queue: Queues created - please start the topic producer +Subscribing local queue 'local_europe' to europe-' +Subscribing local queue 'local_news' to news-' +Subscribing local queue 'local_usa' to usa-' +Subscribing local queue 'local_weather' to weather-' That's all, folks! That's all, folks! That's all, folks! diff --git a/cpp/examples/examples/pub-sub/verify_python_cpp.in b/cpp/examples/examples/pub-sub/verify_python_cpp.in index 97fccf0a32..52e8db9d72 100644 --- a/cpp/examples/examples/pub-sub/verify_python_cpp.in +++ b/cpp/examples/examples/pub-sub/verify_python_cpp.in @@ -5,50 +5,50 @@ Declaring queue: news Declaring queue: usa Declaring queue: weather Listening for messages ... -Message: message 0 from europe -Message: message 0 from europe -Message: message 0 from news -Message: message 0 from news -Message: message 0 from usa -Message: message 0 from usa -Message: message 0 from weather -Message: message 0 from weather -Message: message 1 from europe -Message: message 1 from europe -Message: message 1 from news -Message: message 1 from news -Message: message 1 from usa -Message: message 1 from usa -Message: message 1 from weather -Message: message 1 from weather -Message: message 2 from europe -Message: message 2 from europe -Message: message 2 from news -Message: message 2 from news -Message: message 2 from usa -Message: message 2 from usa -Message: message 2 from weather -Message: message 2 from weather -Message: message 3 from europe -Message: message 3 from europe -Message: message 3 from news -Message: message 3 from news -Message: message 3 from usa -Message: message 3 from usa -Message: message 3 from weather -Message: message 3 from weather -Message: message 4 from europe -Message: message 4 from europe -Message: message 4 from news -Message: message 4 from news -Message: message 4 from usa -Message: message 4 from usa -Message: message 4 from weather -Message: message 4 from weather +Message: europe.news 0 from europe +Message: europe.news 0 from news +Message: europe.news 1 from europe +Message: europe.news 1 from news +Message: europe.news 2 from europe +Message: europe.news 2 from news +Message: europe.news 3 from europe +Message: europe.news 3 from news +Message: europe.news 4 from europe +Message: europe.news 4 from news +Message: europe.weather 0 from europe +Message: europe.weather 0 from weather +Message: europe.weather 1 from europe +Message: europe.weather 1 from weather +Message: europe.weather 2 from europe +Message: europe.weather 2 from weather +Message: europe.weather 3 from europe +Message: europe.weather 3 from weather +Message: europe.weather 4 from europe +Message: europe.weather 4 from weather Message: That's all, folks! from europe Message: That's all, folks! from news Message: That's all, folks! from usa Message: That's all, folks! from weather +Message: usa.news 0 from news +Message: usa.news 0 from usa +Message: usa.news 1 from news +Message: usa.news 1 from usa +Message: usa.news 2 from news +Message: usa.news 2 from usa +Message: usa.news 3 from news +Message: usa.news 3 from usa +Message: usa.news 4 from news +Message: usa.news 4 from usa +Message: usa.weather 0 from usa +Message: usa.weather 0 from weather +Message: usa.weather 1 from usa +Message: usa.weather 1 from weather +Message: usa.weather 2 from usa +Message: usa.weather 2 from weather +Message: usa.weather 3 from usa +Message: usa.weather 3 from weather +Message: usa.weather 4 from usa +Message: usa.weather 4 from weather Shutting down listener for europe Shutting down listener for news Shutting down listener for usa diff --git a/cpp/examples/examples/request-response/client.cpp b/cpp/examples/examples/request-response/client.cpp index 8cec16a461..073af596bf 100644 --- a/cpp/examples/examples/request-response/client.cpp +++ b/cpp/examples/examples/request-response/client.cpp @@ -61,9 +61,9 @@ private: int counter; public: Listener(Session& session, string destination_name): + session(session), destination_name(destination_name), dispatcher(session), - session(session), counter(0) {}; diff --git a/cpp/examples/examples/request-response/verify_python_cpp.in b/cpp/examples/examples/request-response/verify_python_cpp.in index 41ec500774..ba9e01f465 100644 --- a/cpp/examples/examples/request-response/verify_python_cpp.in +++ b/cpp/examples/examples/request-response/verify_python_cpp.in @@ -3,7 +3,7 @@ Request: Twas brilling, and the slithy toves Request: Did gyre and gimble in the wabe. Request: All mimsy were the borogroves, Request: And the mome raths outgrabe. -Messages queue: ReplyTo: +Messages on queue: reply_to: Response: TWAS BRILLING, AND THE SLITHY TOVES Response: DID GYRE AND GIMBLE IN THE WABE. Response: ALL MIMSY WERE THE BOROGROVES, @@ -12,7 +12,7 @@ No more messages! ==== server.out | remove_uuid Activating request queue listener for: request Waiting for requests -Request: Twas brilling, and the slithy toves (ReplyTo:) -Request: Did gyre and gimble in the wabe. (ReplyTo:) -Request: All mimsy were the borogroves, (ReplyTo:) -Request: And the mome raths outgrabe. (ReplyTo:) +Request: Twas brilling, and the slithy toves (reply_to:) +Request: Did gyre and gimble in the wabe. (reply_to:) +Request: All mimsy were the borogroves, (reply_to:) +Request: And the mome raths outgrabe. (reply_to:) diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_python.in b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_python.in index 2e96f03318..0089e55c16 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_python.in +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/fanout/verify_java_python.in @@ -15,47 +15,41 @@ Producer: Sending message: 10 Producer: Closing connection Producer: Closing JNDI context ==== fanout_consumer.py.out | remove_uuid -Messages queue: Subscribed to queue -Response: Message 1 -Response: Message 2 -Response: Message 3 -Response: Message 4 -Response: Message 5 -Response: Message 6 -Response: Message 7 -Response: Message 8 -Response: Message 9 -Response: Message 10 -Response: That's all, folks! -No more messages! +Message 1 +Message 2 +Message 3 +Message 4 +Message 5 +Message 6 +Message 7 +Message 8 +Message 9 +Message 10 +That's all, folks! ==== fanout_consumer.pyX.out | remove_uuid -Messages queue: Subscribed to queue -Response: Message 1 -Response: Message 2 -Response: Message 3 -Response: Message 4 -Response: Message 5 -Response: Message 6 -Response: Message 7 -Response: Message 8 -Response: Message 9 -Response: Message 10 -Response: That's all, folks! -No more messages! +Message 1 +Message 2 +Message 3 +Message 4 +Message 5 +Message 6 +Message 7 +Message 8 +Message 9 +Message 10 +That's all, folks! ==== fanout_consumer.pyXX.out | remove_uuid -Messages queue: Subscribed to queue -Response: Message 1 -Response: Message 2 -Response: Message 3 -Response: Message 4 -Response: Message 5 -Response: Message 6 -Response: Message 7 -Response: Message 8 -Response: Message 9 -Response: Message 10 -Response: That's all, folks! -No more messages! +Message 1 +Message 2 +Message 3 +Message 4 +Message 5 +Message 6 +Message 7 +Message 8 +Message 9 +Message 10 +That's all, folks! diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_python.in b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_python.in index 1a8962515d..92184201d0 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_python.in +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_java_python.in @@ -80,11 +80,15 @@ message 6 message 6 message 6 message 6 -Messages queue: europe -Messages queue: news -Messages queue: usa -Messages queue: weather +Messages on 'europe' queue: +Messages on 'news' queue: +Messages on 'usa' queue: +Messages on 'weather' queue: Queues created - please start the topic producer +Subscribing local queue 'local_europe' to europe-' +Subscribing local queue 'local_news' to news-' +Subscribing local queue 'local_usa' to usa-' +Subscribing local queue 'local_weather' to weather-' That's all, folks! That's all, folks! That's all, folks! diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_python_java.in b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_python_java.in index 507a51f78c..68b96cba2b 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_python_java.in +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/pubsub/verify_python_java.in @@ -7,46 +7,46 @@ Listener: Creating a Message Subscriber for topic news Listener: Creating a Message Subscriber for topic usa Listener: Creating a Message Subscriber for topic weather Listener: Creating a non-transacted, auto-acknowledged session -Listener: Received message for topic: europe: message 0 -Listener: Received message for topic: europe: message 0 -Listener: Received message for topic: europe: message 1 -Listener: Received message for topic: europe: message 1 -Listener: Received message for topic: europe: message 2 -Listener: Received message for topic: europe: message 2 -Listener: Received message for topic: europe: message 3 -Listener: Received message for topic: europe: message 3 -Listener: Received message for topic: europe: message 4 -Listener: Received message for topic: europe: message 4 -Listener: Received message for topic: news: message 0 -Listener: Received message for topic: news: message 0 -Listener: Received message for topic: news: message 1 -Listener: Received message for topic: news: message 1 -Listener: Received message for topic: news: message 2 -Listener: Received message for topic: news: message 2 -Listener: Received message for topic: news: message 3 -Listener: Received message for topic: news: message 3 -Listener: Received message for topic: news: message 4 -Listener: Received message for topic: news: message 4 -Listener: Received message for topic: usa: message 0 -Listener: Received message for topic: usa: message 0 -Listener: Received message for topic: usa: message 1 -Listener: Received message for topic: usa: message 1 -Listener: Received message for topic: usa: message 2 -Listener: Received message for topic: usa: message 2 -Listener: Received message for topic: usa: message 3 -Listener: Received message for topic: usa: message 3 -Listener: Received message for topic: usa: message 4 -Listener: Received message for topic: usa: message 4 -Listener: Received message for topic: weather: message 0 -Listener: Received message for topic: weather: message 0 -Listener: Received message for topic: weather: message 1 -Listener: Received message for topic: weather: message 1 -Listener: Received message for topic: weather: message 2 -Listener: Received message for topic: weather: message 2 -Listener: Received message for topic: weather: message 3 -Listener: Received message for topic: weather: message 3 -Listener: Received message for topic: weather: message 4 -Listener: Received message for topic: weather: message 4 +Listener: Received message for topic: europe: europe.news 0 +Listener: Received message for topic: europe: europe.news 1 +Listener: Received message for topic: europe: europe.news 2 +Listener: Received message for topic: europe: europe.news 3 +Listener: Received message for topic: europe: europe.news 4 +Listener: Received message for topic: europe: europe.weather 0 +Listener: Received message for topic: europe: europe.weather 1 +Listener: Received message for topic: europe: europe.weather 2 +Listener: Received message for topic: europe: europe.weather 3 +Listener: Received message for topic: europe: europe.weather 4 +Listener: Received message for topic: news: europe.news 0 +Listener: Received message for topic: news: europe.news 1 +Listener: Received message for topic: news: europe.news 2 +Listener: Received message for topic: news: europe.news 3 +Listener: Received message for topic: news: europe.news 4 +Listener: Received message for topic: news: usa.news 0 +Listener: Received message for topic: news: usa.news 1 +Listener: Received message for topic: news: usa.news 2 +Listener: Received message for topic: news: usa.news 3 +Listener: Received message for topic: news: usa.news 4 +Listener: Received message for topic: usa: usa.news 0 +Listener: Received message for topic: usa: usa.news 1 +Listener: Received message for topic: usa: usa.news 2 +Listener: Received message for topic: usa: usa.news 3 +Listener: Received message for topic: usa: usa.news 4 +Listener: Received message for topic: usa: usa.weather 0 +Listener: Received message for topic: usa: usa.weather 1 +Listener: Received message for topic: usa: usa.weather 2 +Listener: Received message for topic: usa: usa.weather 3 +Listener: Received message for topic: usa: usa.weather 4 +Listener: Received message for topic: weather: europe.weather 0 +Listener: Received message for topic: weather: europe.weather 1 +Listener: Received message for topic: weather: europe.weather 2 +Listener: Received message for topic: weather: europe.weather 3 +Listener: Received message for topic: weather: europe.weather 4 +Listener: Received message for topic: weather: usa.weather 0 +Listener: Received message for topic: weather: usa.weather 1 +Listener: Received message for topic: weather: usa.weather 2 +Listener: Received message for topic: weather: usa.weather 3 +Listener: Received message for topic: weather: usa.weather 4 Listener: Setting an ExceptionListener on the connection as sample uses a TopicSubscriber Listener: Shutting down listener for europe Listener: Shutting down listener for news diff --git a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_python.in b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_python.in index 25285eb2a3..ead7a5747b 100644 --- a/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_python.in +++ b/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/requestResponse/verify_java_python.in @@ -3,7 +3,7 @@ Request: Twas brilling, and the slithy toves Request: Did gyre and gimble in the wabe. Request: All mimsy were the borogroves, Request: And the mome raths outgrabe. -Messages queue: ReplyTo: +Messages on queue: reply_to: Response: TWAS BRILLING, AND THE SLITHY TOVES Response: DID GYRE AND GIMBLE IN THE WABE. Response: ALL MIMSY WERE THE BOROGROVES, diff --git a/python/examples/direct/declare_queues.py b/python/examples/direct/declare_queues.py index 7041ce2f24..deea0a3ccc 100755 --- a/python/examples/direct/declare_queues.py +++ b/python/examples/direct/declare_queues.py @@ -8,34 +8,47 @@ sent to the queue named "message_queue". """ +# Common includes + import qpid import sys import os -from random import randint from qpid.util import connect from qpid.connection import Connection +from qpid.datatypes import Message, RangedSet, uuid4 from qpid.queue import Empty #----- Initialization ----------------------------------- # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] except KeyError: - amqp_spec="/usr/share/amqp/amqp.0-10.xml" + amqp_spec="/usr/share/amqp/amqp.0-10.xml" # Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() - -session = conn.session(str(randint(1,64*1024))) +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) #----- Create a queue ------------------------------------- diff --git a/python/examples/direct/direct_consumer.py b/python/examples/direct/direct_consumer.py index 91d85cee1a..f2018bbbb8 100755 --- a/python/examples/direct/direct_consumer.py +++ b/python/examples/direct/direct_consumer.py @@ -12,7 +12,7 @@ import os from random import randint from qpid.util import connect from qpid.connection import Connection -from qpid.datatypes import Message, RangedSet +from qpid.datatypes import Message, RangedSet, uuid4 from qpid.queue import Empty @@ -20,11 +20,22 @@ from qpid.queue import Empty # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] @@ -32,10 +43,10 @@ except KeyError: amqp_spec="/usr/share/amqp/amqp.0-10.xml" # Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() - -session = conn.session(str(randint(1,64*1024))) +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) #----- Read from queue -------------------------------------------- @@ -44,16 +55,17 @@ session = conn.session(str(randint(1,64*1024))) # The consumer tag identifies the client-side queue. -consumer_tag = "consumer1" -queue = session.incoming(consumer_tag) +local_queue_name = "local_queue" +queue = session.incoming(local_queue_name) -# Call message_consume() to tell the broker to deliver messages +# Call message_subscribe() to tell the broker to deliver messages # from the AMQP queue to this local client queue. The broker will -# start delivering messages as soon as message_consume() is called. +# start delivering messages as soon as credit is allocated using +# session.message_flow(). -session.message_subscribe(queue="message_queue", destination=consumer_tag) -session.message_flow(consumer_tag, 0, 0xFFFFFFFF) # Kill these? -session.message_flow(consumer_tag, 1, 0xFFFFFFFF) # Kill these? +session.message_subscribe(queue="message_queue", destination=local_queue_name) +session.message_flow(local_queue_name, session.credit_unit.message, 0xFFFFFFFF) +session.message_flow(local_queue_name, session.credit_unit.byte, 0xFFFFFFFF) # Initialize 'final' and 'content', variables used to identify the last message. @@ -67,15 +79,6 @@ while content != final: session.message_accept(RangedSet(message.id)) print content -# Messages are not removed from the queue until they are -# acknowledged. Using cumulative=True, all messages from the session -# up to and including the one identified by the delivery tag are -# acknowledged. This is more efficient, because there are fewer -# network round-trips. - -#message.complete(cumulative=True) -# ? Is there an equivakent to the above in the new API ? - #----- Cleanup ------------------------------------------------ # Clean up before exiting so there are no open threads. diff --git a/python/examples/direct/direct_producer.py b/python/examples/direct/direct_producer.py index 7c4e30d96e..8f6a91ba18 100755 --- a/python/examples/direct/direct_producer.py +++ b/python/examples/direct/direct_producer.py @@ -9,21 +9,33 @@ import qpid import sys import os -from random import randint from qpid.util import connect from qpid.connection import Connection from qpid.datatypes import Message +from qpid.datatypes import uuid4 from qpid.queue import Empty + #----- Initialization ----------------------------------- # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] @@ -31,10 +43,10 @@ except KeyError: amqp_spec="/usr/share/amqp/amqp.0-10.xml" # Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() - -session = conn.session(str(randint(1,64*1024))) +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) #----- Publish some messages ------------------------------ @@ -42,9 +54,9 @@ session = conn.session(str(randint(1,64*1024))) props = session.delivery_properties(routing_key="routing_key") for i in range(10): - session.message_transfer("amq.direct",None, None, Message(props,"message " + str(i))) + session.message_transfer(destination="amq.direct", message=Message(props,"message " + str(i))) -session.message_transfer("amq.direct",None,None, Message(props,"That's all, folks!")) +session.message_transfer(destination="amq.direct", message=Message(props,"That's all, folks!")) #----- Cleanup -------------------------------------------- diff --git a/python/examples/direct/listener.py b/python/examples/direct/listener.py index aa60b1c501..c18ef47fb7 100755 --- a/python/examples/direct/listener.py +++ b/python/examples/direct/listener.py @@ -7,14 +7,18 @@ as a message listener. """ +# Common includes + import qpid import sys import os -from random import randint from qpid.util import connect from qpid.connection import Connection -from qpid.datatypes import Message, RangedSet -from qpid.queue import Empty +from qpid.datatypes import Message, RangedSet, uuid4 +from qpid.queue import Empty + +# Includes specific to this example + from time import sleep @@ -33,23 +37,26 @@ class Receiver: if content == "That's all, folks!": self.finalReceived = True - # Messages are not removed from the queue until they are - # acknowledged. Using cumulative=True, all messages from the session - # up to and including the one identified by the delivery tag are - # acknowledged. This is more efficient, because there are fewer - # network round-trips. - #message.complete(cumulative=True) - - #----- Initialization -------------------------------------- # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] @@ -57,33 +64,33 @@ except KeyError: amqp_spec="/usr/share/amqp/amqp.0-10.xml" # Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() - -session = conn.session(str(randint(1,64*1024))) +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) #----- Read from queue -------------------------------------------- # Now let's create a local client queue and tell it to read # incoming messages. -# The consumer tag identifies the client-side queue. +# The local_queue_name identifies the client-side queue. -consumer_tag = "consumer1" -queue = session.incoming(consumer_tag) +local_queue_name = "local_queue" +queue = session.incoming(local_queue_name) # Call message_subscribe() to tell the broker to deliver messages # from the AMQP queue to this local client queue. The broker will # start delivering messages as soon as message_subscribe() is called. -session.message_subscribe(queue="message_queue", destination=consumer_tag) -session.message_flow(consumer_tag, 0, 0xFFFFFFFF) # Kill these? -session.message_flow(consumer_tag, 1, 0xFFFFFFFF) # Kill these? +session.message_subscribe(queue="message_queue", destination=local_queue_name) +session.message_flow(local_queue_name, session.credit_unit.message, 0xFFFFFFFF) +session.message_flow(local_queue_name, session.credit_unit.byte, 0xFFFFFFFF) -receiver = Receiver () +receiver = Receiver() queue.listen (receiver.Handler) -while not receiver.isFinal (): +while not receiver.isFinal() : sleep (1) diff --git a/python/examples/fanout/fanout_consumer.py b/python/examples/fanout/fanout_consumer.py index b82d8045ff..21fc5e8f16 100755 --- a/python/examples/fanout/fanout_consumer.py +++ b/python/examples/fanout/fanout_consumer.py @@ -13,61 +13,27 @@ from qpid.connection import Connection from qpid.datatypes import Message, RangedSet, uuid4 from qpid.queue import Empty -#----- Functions ------------------------------------------- - -def dump_queue(session, queue_name): - - print "Messages queue: " + queue_name - - consumer_tag = queue_name # Use the queue name as the consumer tag - need a unique tag - queue = session.incoming(queue_name) - - # Call message_subscribe() to tell the broker to deliver messages - # from the AMQP queue to a local client queue. The broker will - # start delivering messages as soon as message_subscribe() is called. - - session.message_subscribe(queue=queue_name, destination=consumer_tag) - session.message_flow(consumer_tag, 0, 0xFFFFFFFF) - session.message_flow(consumer_tag, 1, 0xFFFFFFFF) - - print "Subscribed to queue " + queue_name - sys.stdout.flush() - - message = 0 - - while True: - try: - message = queue.get(timeout=10) - content = message.body - session.message_accept(RangedSet(message.id)) - print "Response: " + content - except Empty: - print "No more messages!" - break - except: - print "Unexpected exception!" - break - - - # Messages are not removed from the queue until they - # are acknowledged. Using cumulative=True, all messages - # in the session up to and including the one identified - # by the delivery tag are acknowledged. This is more efficient, - # because there are fewer network round-trips. - - #if message != 0: - # message.complete(cumulative=True) - - #----- Initialization -------------------------------------- + # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] @@ -75,25 +41,48 @@ except KeyError: amqp_spec="/usr/share/amqp/amqp.0-10.xml" # Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) + + +#----- Main Body ------------------------------------------- + +# Create a server-side queue and route messages to it. +# The server-side queue must have a unique name. Use the +# session id for that. +server_queue_name = session.name +session.queue_declare(queue=server_queue_name) +session.exchange_bind(queue=server_queue_name, exchange="amq.fanout") + +# Create a local queue to receive messages from the server-side +# queue. +local_queue_name = "local_queue" +local_queue = session.incoming(local_queue_name) + +# Call message_consume() to tell the server to deliver messages +# from the AMQP queue to this local client queue. -session_id = str(uuid4()) -session = conn.session(session_id) +session.message_subscribe(queue=server_queue_name, destination=local_queue_name) +session.message_flow(local_queue_name, session.credit_unit.message, 0xFFFFFFFF) +session.message_flow(local_queue_name, session.credit_unit.byte, 0xFFFFFFFF) -#----- Main Body -- ---------------------------------------- +print "Subscribed to queue " + server_queue_name +sys.stdout.flush() -# Make a unique queue name for my queue from the session ID. -my_queue = session_id -session.queue_declare(queue=my_queue) +# Initialize 'final' and 'content', variables used to identify the last message. +final = "That's all, folks!" # In a message body, signals the last message +content = "" # Content of the last message read -# Bind my queue to the fanout exchange. No routing key is required -# the fanout exchange copies messages unconditionally to every -# bound queue -session.exchange_bind(queue=my_queue, exchange="amq.fanout") +# Read the messages - acknowledge each one +message = None +while content != final: + message = local_queue.get(timeout=10) + content = message.body + session.message_accept(RangedSet(message.id)) + print content -# Dump the messages on the queue. -dump_queue(session, my_queue) #----- Cleanup ------------------------------------------------ diff --git a/python/examples/fanout/fanout_producer.py b/python/examples/fanout/fanout_producer.py index 1b5ea6995e..43d6a94c3d 100755 --- a/python/examples/fanout/fanout_producer.py +++ b/python/examples/fanout/fanout_producer.py @@ -8,21 +8,31 @@ import qpid import sys import os -from random import randint from qpid.util import connect from qpid.connection import Connection -from qpid.datatypes import Message +from qpid.datatypes import Message, uuid4 from qpid.queue import Empty #----- Initialization ----------------------------------- # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] @@ -30,19 +40,22 @@ except KeyError: amqp_spec="/usr/share/amqp/amqp.0-10.xml" # Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) -session = conn.session(str(randint(1,64*1024))) #----- Publish some messages ------------------------------ # Create some messages and put them on the broker. +delivery_properties = session.delivery_properties(routing_key="routing_key") + for i in range(10): - session.message_transfer("amq.fanout", None, None ,Message("message " + str(i))) + session.message_transfer(destination="amq.fanout", message=Message(delivery_properties,"message " + str(i))) -session.message_transfer("amq.fanout", None, None, Message("That's all, folks!")) +session.message_transfer(destination="amq.fanout", message=Message(delivery_properties, "That's all, folks!")) #----- Cleanup -------------------------------------------- diff --git a/python/examples/fanout/listener.py b/python/examples/fanout/listener.py index 8997c3698f..50cd06d2a5 100755 --- a/python/examples/fanout/listener.py +++ b/python/examples/fanout/listener.py @@ -8,11 +8,15 @@ import qpid import sys -from qpid.client import Client -from qpid.content import Content -from qpid.queue import Empty -from time import sleep +import os +from qpid.util import connect +from qpid.connection import Connection +from qpid.datatypes import Message, RangedSet, uuid4 +from qpid.queue import Empty + +# +from time import sleep #----- Message Receive Handler ----------------------------- class Receiver: @@ -23,57 +27,76 @@ class Receiver: return self.finalReceived def Handler (self, message): - content = message.content.body + content = message.body + session.message_accept(RangedSet(message.id)) print content if content == "That's all, folks!": self.finalReceived = True - # Messages are not removed from the queue until they are - # acknowledged. Using cumulative=True, all messages from the session - # up to and including the one identified by the delivery tag are - # acknowledged. This is more efficient, because there are fewer - # network round-trips. - message.complete(cumulative=True) - #----- Initialization -------------------------------------- # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 -amqp_spec="/usr/share/amqp/amqp.0-10-preview.xml" +host="127.0.0.1" +port=5672 user="guest" password="guest" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" -# Create a client and log in to it. +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. -client = Client(host, port, qpid.spec.load(amqp_spec)) -client.start({"LOGIN": user, "PASSWORD": password}) +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) -session = client.session() -session.session_open() +try: + amqp_spec = os.environ["AMQP_SPEC"] +except KeyError: + amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# Create a connection. +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) #----- Read from queue -------------------------------------------- -# Now let's create a local client queue and tell it to read -# incoming messages. +# Create a server-side queue and route messages to it. +# The server-side queue must have a unique name. Use the +# session id for that. + +server_queue_name = session.name +session.queue_declare(queue=server_queue_name) +session.exchange_bind(queue=server_queue_name, exchange="amq.fanout") + +# Create a local queue to receive messages from the server-side +# queue. +local_queue_name = "local_queue" +local_queue = session.incoming(local_queue_name) + -# The consumer tag identifies the client-side queue. +# The local queue name identifies the client-side queue. -consumer_tag = "consumer1" -queue = client.queue(consumer_tag) +local_queue_name = "local_queue" +local_queue = session.incoming(local_queue_name) # Call message_subscribe() to tell the broker to deliver messages # from the AMQP queue to this local client queue. The broker will # start delivering messages as soon as message_subscribe() is called. -session.message_subscribe(queue="message_queue", destination=consumer_tag) -session.message_flow(consumer_tag, 0, 0xFFFFFFFF) # Kill these? -session.message_flow(consumer_tag, 1, 0xFFFFFFFF) # Kill these? +session.message_subscribe(queue=server_queue_name, destination=local_queue_name) +session.message_flow(local_queue_name, session.credit_unit.message, 0xFFFFFFFF) +session.message_flow(local_queue_name, session.credit_unit.byte, 0xFFFFFFFF) receiver = Receiver () -queue.listen (receiver.Handler) +local_queue.listen (receiver.Handler) while not receiver.isFinal (): sleep (1) @@ -84,4 +107,4 @@ while not receiver.isFinal (): # Clean up before exiting so there are no open threads. # -session.session_close() +session.close() diff --git a/python/examples/fanout/verify.in b/python/examples/fanout/verify.in index 30dfeb9e69..d4b8670de9 100644 --- a/python/examples/fanout/verify.in +++ b/python/examples/fanout/verify.in @@ -1,31 +1,27 @@ ==== fanout_producer.py.out ==== fanout_consumer.py.out | remove_uuid -Messages queue: Subscribed to queue -Response: message 0 -Response: message 1 -Response: message 2 -Response: message 3 -Response: message 4 -Response: message 5 -Response: message 6 -Response: message 7 -Response: message 8 -Response: message 9 -Response: That's all, folks! -No more messages! +message 0 +message 1 +message 2 +message 3 +message 4 +message 5 +message 6 +message 7 +message 8 +message 9 +That's all, folks! ==== fanout_consumer.pyX.out | remove_uuid -Messages queue: Subscribed to queue -Response: message 0 -Response: message 1 -Response: message 2 -Response: message 3 -Response: message 4 -Response: message 5 -Response: message 6 -Response: message 7 -Response: message 8 -Response: message 9 -Response: That's all, folks! -No more messages! +message 0 +message 1 +message 2 +message 3 +message 4 +message 5 +message 6 +message 7 +message 8 +message 9 +That's all, folks! diff --git a/python/examples/pubsub/topic_publisher.py b/python/examples/pubsub/topic_publisher.py index b79896eaf6..64e5a99924 100755 --- a/python/examples/pubsub/topic_publisher.py +++ b/python/examples/pubsub/topic_publisher.py @@ -10,20 +10,38 @@ import qpid import sys import os -from random import randint from qpid.util import connect from qpid.connection import Connection -from qpid.datatypes import Message +from qpid.datatypes import Message, RangedSet, uuid4 from qpid.queue import Empty +#----- Functions ---------------------------------------- + +def send_msg(routing_key): + props = session.delivery_properties(routing_key=routing_key) + for i in range(5): + session.message_transfer(destination="amq.topic", message=Message(props,routing_key + " " + str(i))) + #----- Initialization ----------------------------------- -# Set parameters for login. -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 +# Set parameters for login + +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] @@ -31,10 +49,10 @@ except KeyError: amqp_spec="/usr/share/amqp/amqp.0-10.xml" # Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() - -session = conn.session(str(randint(1,64*1024))) +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) #----- Publish some messages ------------------------------ @@ -42,11 +60,6 @@ session = conn.session(str(randint(1,64*1024))) # topic exchange. The routing keys are "usa.news", "usa.weather", # "europe.news", and "europe.weather". -def send_msg(routing_key): - props = session.delivery_properties(routing_key=routing_key) - for i in range(5): - session.message_transfer("amq.topic", None, None, Message(props,"message " + str(i))) - # usa.news send_msg("usa.news") @@ -61,7 +74,7 @@ send_msg("europe.weather") # Signal termination props = session.delivery_properties(routing_key="control") -session.message_transfer("amq.topic",None, None, Message(props,"That's all, folks!")) +session.message_transfer(destination="amq.topic", message=Message(props,"That's all, folks!")) #----- Cleanup -------------------------------------------- diff --git a/python/examples/pubsub/topic_subscriber.py b/python/examples/pubsub/topic_subscriber.py index 6908be5471..3c4a8d8d0c 100755 --- a/python/examples/pubsub/topic_subscriber.py +++ b/python/examples/pubsub/topic_subscriber.py @@ -16,20 +16,7 @@ from qpid.queue import Empty #----- Functions ------------------------------------------- -def dump_queue(queue_name): - - print "Messages queue: " + queue_name - - consumer_tag = queue_name # Use the queue name as the consumer tag - need a unique tag - queue = session.incoming(consumer_tag) - - # Call message_subscribe() to tell the broker to deliver messages - # from the AMQP queue to a local client queue. The broker will - # start delivering messages as soon as message_subscribe() is called. - - session.message_subscribe(queue=queue_name, destination=consumer_tag) - session.message_flow(consumer_tag, 0, 0xFFFFFFFF) - session.message_flow(consumer_tag, 1, 0xFFFFFFFF) +def dump_queue(queue): content = "" # Content of the last message read final = "That's all, folks!" # In a message body, signals the last message @@ -37,36 +24,48 @@ def dump_queue(queue_name): while content != final: try: - message = queue.get() + message = queue.get(timeout=10) content = message.body session.message_accept(RangedSet(message.id)) print content except Empty: - #if message != 0: - # message.complete(cumulative=True) print "No more messages!" return - # Messages are not removed from the queue until they - # are acknowledged. Using multiple=True, all messages - # in the channel up to and including the one identified - # by the delivery tag are acknowledged. This is more efficient, - # because there are fewer network round-trips. - #if message != 0: - # message.complete(cumulative=True) +def subscribe_queue(server_queue_name, local_queue_name): + print "Subscribing local queue '" + local_queue_name + "' to " + server_queue_name + "'" + + queue = session.incoming(local_queue_name) + + session.message_subscribe(queue=server_queue_name, destination=local_queue_name) + session.message_flow(local_queue_name, session.credit_unit.message, 0xFFFFFFFF) + session.message_flow(local_queue_name, session.credit_unit.byte, 0xFFFFFFFF) + + return queue #----- Initialization -------------------------------------- # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] @@ -74,19 +73,19 @@ except KeyError: amqp_spec="/usr/share/amqp/amqp.0-10.xml" # Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() - -session_id = str(uuid4()) -session = conn.session(session_id) +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) #----- Main Body -- ---------------------------------------- +# declare queues on the server -news = "news" + session_id -weather = "weather" + session_id -usa = "usa" + session_id -europe = "europe" + session_id +news = "news-" + session.name +weather = "weather-" + session.name +usa = "usa-" + session.name +europe = "europe-" + session.name session.queue_declare(queue=news, exclusive=True) session.queue_declare(queue=weather, exclusive=True) @@ -115,12 +114,31 @@ session.exchange_bind(exchange="amq.topic", queue=europe, binding_key="control") print "Queues created - please start the topic producer" sys.stdout.flush() +# Subscribe local queues to server queues + +local_news = "local_news" +local_weather = "local_weather" +local_usa = "local_usa" +local_europe = "local_europe" + +local_news_queue = subscribe_queue(news, local_news) +local_weather_queue = subscribe_queue(weather, local_weather) +local_usa_queue = subscribe_queue(usa, local_usa) +local_europe_queue = subscribe_queue(europe, local_europe) + # Call dump_queue to print messages from each queue -dump_queue(news) -dump_queue(weather) -dump_queue(usa) -dump_queue(europe) +print "Messages on 'news' queue:" +dump_queue(local_news_queue) + +print "Messages on 'weather' queue:" +dump_queue(local_weather_queue) + +print "Messages on 'usa' queue:" +dump_queue(local_usa_queue) + +print "Messages on 'europe' queue:" +dump_queue(local_europe_queue) #----- Cleanup ------------------------------------------------ diff --git a/python/examples/pubsub/verify.in b/python/examples/pubsub/verify.in index 2f6da09ec5..1b74acd832 100644 --- a/python/examples/pubsub/verify.in +++ b/python/examples/pubsub/verify.in @@ -1,51 +1,55 @@ ==== topic_publisher.py.out ==== topic_subscriber.py.out | remove_uuid | sort -message 0 -message 0 -message 0 -message 0 -message 0 -message 0 -message 0 -message 0 -message 1 -message 1 -message 1 -message 1 -message 1 -message 1 -message 1 -message 1 -message 2 -message 2 -message 2 -message 2 -message 2 -message 2 -message 2 -message 2 -message 3 -message 3 -message 3 -message 3 -message 3 -message 3 -message 3 -message 3 -message 4 -message 4 -message 4 -message 4 -message 4 -message 4 -message 4 -message 4 -Messages queue: europe -Messages queue: news -Messages queue: usa -Messages queue: weather +europe.news 0 +europe.news 0 +europe.news 1 +europe.news 1 +europe.news 2 +europe.news 2 +europe.news 3 +europe.news 3 +europe.news 4 +europe.news 4 +europe.weather 0 +europe.weather 0 +europe.weather 1 +europe.weather 1 +europe.weather 2 +europe.weather 2 +europe.weather 3 +europe.weather 3 +europe.weather 4 +europe.weather 4 +Messages on 'europe' queue: +Messages on 'news' queue: +Messages on 'usa' queue: +Messages on 'weather' queue: Queues created - please start the topic producer +Subscribing local queue 'local_europe' to europe-' +Subscribing local queue 'local_news' to news-' +Subscribing local queue 'local_usa' to usa-' +Subscribing local queue 'local_weather' to weather-' That's all, folks! That's all, folks! That's all, folks! That's all, folks! +usa.news 0 +usa.news 0 +usa.news 1 +usa.news 1 +usa.news 2 +usa.news 2 +usa.news 3 +usa.news 3 +usa.news 4 +usa.news 4 +usa.weather 0 +usa.weather 0 +usa.weather 1 +usa.weather 1 +usa.weather 2 +usa.weather 2 +usa.weather 3 +usa.weather 3 +usa.weather 4 +usa.weather 4 diff --git a/python/examples/request-response/client.py b/python/examples/request-response/client.py index 8f7d430d1b..0fcd256d49 100755 --- a/python/examples/request-response/client.py +++ b/python/examples/request-response/client.py @@ -18,18 +18,7 @@ from qpid.queue import Empty def dump_queue(queue_name): - print "Messages queue: " + queue_name - - consumer_tag = queue_name # Use the queue name as the consumer tag - need a unique tag - queue = session.incoming(consumer_tag) - - # Call message_subscribe() to tell the broker to deliver messages - # from the AMQP queue to a local client queue. The broker will - # start delivering messages as soon as message_subscribe() is called. - - session.message_subscribe(queue=queue_name, destination=consumer_tag) - session.message_flow(consumer_tag, 0, 0xFFFFFFFF) - session.message_flow(consumer_tag, 1, 0xFFFFFFFF) + print "Messages on queue: " + queue_name message = 0 @@ -47,25 +36,27 @@ def dump_queue(queue_name): break - # Messages are not removed from the queue until they - # are acknowledged. Using cumulative=True, all messages - # in the session up to and including the one identified - # by the delivery tag are acknowledged. This is more efficient, - # because there are fewer network round-trips. - - #if message != 0: - # message.complete(cumulative=True) - - #----- Initialization -------------------------------------- + # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] @@ -73,11 +64,11 @@ except KeyError: amqp_spec="/usr/share/amqp/amqp.0-10.xml" # Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) -session_id = str(uuid4()) -session = conn.session(session_id) #----- Main Body -- ---------------------------------------- @@ -85,9 +76,23 @@ session = conn.session(session_id) # same string as the name of the queue and the name of the routing # key. -replyTo = "ReplyTo:" + session_id -session.queue_declare(queue=replyTo, exclusive=True) -session.exchange_bind(exchange="amq.direct", queue=replyTo, binding_key=replyTo) +reply_to = "reply_to:" + session.name +session.queue_declare(queue=reply_to, exclusive=True) +session.exchange_bind(exchange="amq.direct", queue=reply_to, binding_key=reply_to) + +# Create a local queue and subscribe it to the response queue + +local_queue_name = "local_queue" +queue = session.incoming(local_queue_name) + +# Call message_subscribe() to tell the broker to deliver messages from +# the server's reply_to queue to our local client queue. The server +# will start delivering messages as soon as message credit is +# available. + +session.message_subscribe(queue=reply_to, destination=local_queue_name) +session.message_flow(local_queue_name, session.credit_unit.message, 0xFFFFFFFF) +session.message_flow(local_queue_name, session.credit_unit.byte, 0xFFFFFFFF) # Send some messages to the server's request queue @@ -96,16 +101,20 @@ lines = ["Twas brilling, and the slithy toves", "All mimsy were the borogroves,", "And the mome raths outgrabe."] -for ln in lines: - print "Request: " + ln - mp = session.message_properties() - mp.reply_to = session.reply_to("amq.direct", replyTo) - dp = session.delivery_properties(routing_key="request") - session.message_transfer("amq.direct", None, None, Message(mp,dp,ln)) +# We will use the same reply_to and routing key +# for each message + +message_properties = session.message_properties() +message_properties.reply_to = session.reply_to("amq.direct", reply_to) +delivery_properties = session.delivery_properties(routing_key="request") + +for line in lines: + print "Request: " + line + session.message_transfer(destination="amq.direct", message=Message(message_properties, delivery_properties, line)) -# Now see what messages the server sent to our replyTo queue +# Now see what messages the server sent to our reply_to queue -dump_queue(replyTo) +dump_queue(reply_to) #----- Cleanup ------------------------------------------------ diff --git a/python/examples/request-response/server.py b/python/examples/request-response/server.py index 4377571248..7b182723b9 100755 --- a/python/examples/request-response/server.py +++ b/python/examples/request-response/server.py @@ -4,6 +4,7 @@ Server for a client/server example """ + import qpid import sys import os @@ -22,31 +23,42 @@ def respond(session, request): message_properties = request.get("message_properties") reply_to = message_properties.reply_to if reply_to == None: - raise Exception("reply to property needs to be there") - - props = session.delivery_properties(routing_key=reply_to["routing_key"]) - session.message_transfer(reply_to["exchange"],None, None, Message(props,request.body.upper())) + raise Exception("This message is missing the 'reply_to' property, which is required") + + props = session.delivery_properties(routing_key=reply_to["routing_key"]) + session.message_transfer(destination=reply_to["exchange"], message=Message(props,request.body.upper())) #----- Initialization -------------------------------------- + # Set parameters for login -host=len(sys.argv) > 1 and sys.argv[1] or "127.0.0.1" -port=len(sys.argv) > 2 and int(sys.argv[2]) or 5672 + +host="127.0.0.1" +port=5672 user="guest" password="guest" -amqp_spec="" +amqp_spec="/usr/share/amqp/amqp.0-10.xml" + +# If an alternate host or port has been specified, use that instead +# (this is used in our unit tests) +# +# If AMQP_SPEC is defined, use it to locate the spec file instead of +# looking for it in the default location. + +if len(sys.argv) > 1 : + host=sys.argv[1] +if len(sys.argv) > 2 : + port=int(sys.argv[2]) try: amqp_spec = os.environ["AMQP_SPEC"] except KeyError: amqp_spec="/usr/share/amqp/amqp.0-10.xml" -# Create a connection. -conn = Connection (connect (host,port), qpid.spec.load(amqp_spec)) -conn.start() - -session_id = str(uuid4()) -session = conn.session(session_id) +socket = connect(host, port) +connection = Connection (sock=socket, spec=qpid.spec.load(amqp_spec)) +connection.start() +session = connection.session(str(uuid4())) #----- Main Body -- ---------------------------------------- @@ -55,11 +67,11 @@ session = conn.session(session_id) session.queue_declare(queue="request", exclusive=True) session.exchange_bind(exchange="amq.direct", queue="request", binding_key="request") -dest = "request_destination" +local_queue_name = "local_queue" -session.message_subscribe(queue="request", destination=dest) -session.message_flow(dest, 0, 0xFFFFFFFF) -session.message_flow(dest, 1, 0xFFFFFFFF) +session.message_subscribe(queue="request", destination=local_queue_name) +session.message_flow(local_queue_name, session.credit_unit.message, 0xFFFFFFFF) +session.message_flow(local_queue_name, session.credit_unit.byte, 0xFFFFFFFF) # Remind the user to start the client program @@ -70,7 +82,7 @@ sys.stdout.flush() # Respond to each request -queue = session.incoming(dest) +queue = session.incoming(local_queue_name) # If we get a message, send it back to the user (as indicated in the # ReplyTo property) diff --git a/python/examples/request-response/verify.in b/python/examples/request-response/verify.in index 8d7f732ec8..6c24366722 100644 --- a/python/examples/request-response/verify.in +++ b/python/examples/request-response/verify.in @@ -3,7 +3,7 @@ Request: Twas brilling, and the slithy toves Request: Did gyre and gimble in the wabe. Request: All mimsy were the borogroves, Request: And the mome raths outgrabe. -Messages queue: ReplyTo: +Messages on queue: reply_to: Response: TWAS BRILLING, AND THE SLITHY TOVES Response: DID GYRE AND GIMBLE IN THE WABE. Response: ALL MIMSY WERE THE BOROGROVES, |