Merge remote branch 'origin/bug23558'

Author: Emile Joubert

dotnet/Receive.cs

@@ -10,10 +10,10 @@ static void Main(string[] args) {
             IConnection connection = factory.CreateConnection();
             IModel channel = connection.CreateModel();
 
-            channel.QueueDeclare("test");
+            channel.QueueDeclare("hello");
 
             QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
-            channel.BasicConsume("test", true, null, consumer);
+            channel.BasicConsume("hello", true, null, consumer);
 
             Console.WriteLine(" [*] Waiting for messages. To exit press CTRL+C");
             while(true) {

dotnet/Send.cs

@@ -9,12 +9,12 @@ static void Main(string[] args) {
             IConnection connection = factory.CreateConnection();
             IModel channel = connection.CreateModel();
 
-            channel.QueueDeclare("test");
+            channel.QueueDeclare("hello");
 
             string message = "Hello World!";
             byte[] body = System.Text.Encoding.UTF8.GetBytes(message);
 
-            channel.BasicPublish("", "test", null, body);
+            channel.BasicPublish("", "hello", null, body);
             Console.WriteLine(" [x] Sent {0}", message);
 
             channel.Close();

java/Recv.java

@@ -10,9 +10,9 @@ public static void main(String[] argv) {
       try {
         conn = new ConnectionFactory().newConnection();
         Channel chan = conn.createChannel();
-        chan.queueDeclare("test", false, false, false, null);
+        chan.queueDeclare("hello", false, false, false, null);
         QueueingConsumer consumer = new QueueingConsumer(chan);
-        chan.basicConsume("test", true, consumer);
+        chan.basicConsume("hello", true, consumer);
         while (true) {
           QueueingConsumer.Delivery delivery = consumer.nextDelivery();
           System.out.println(new String(delivery.getBody()));

java/Send.java

@@ -10,8 +10,8 @@ public static void main(String[] argv) {
       try {
         conn = new ConnectionFactory().newConnection();
         Channel chan = conn.createChannel();
-        chan.queueDeclare("test", false, false, false, null);
-        chan.basicPublish("", "test", null, "Hello World!".getBytes());
+        chan.queueDeclare("hello", false, false, false, null);
+        chan.basicPublish("", "hello", null, "Hello World!".getBytes());
       }
       finally {
         if (conn != null) conn.close();

php/receive.php

@@ -6,15 +6,15 @@
 $channel = $connection->channel();
 
 
-$channel->queue_declare('test', false, false, false, false);
+$channel->queue_declare('hello', false, false, false, false);
 
 echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";
 
 $callback = function($msg) {
   echo " [x] Received ", $msg->body, "\n";
 };
 
-$channel->basic_consume('test', '', false, true, false, false, $callback);
+$channel->basic_consume('hello', '', false, true, false, false, $callback);
 
 while(count($channel->callbacks)) {
     $channel->wait();

php/send.php

@@ -6,10 +6,10 @@
 $channel = $connection->channel();
 
 
-$channel->queue_declare('test', false, false, false, false);
+$channel->queue_declare('hello', false, false, false, false);
 
 $msg = new AMQPMessage('Hello World!');
-$channel->basic_publish($msg, '', 'test');
+$channel->basic_publish($msg, '', 'hello');
 
 echo " [x] Sent 'Hello World!'\n";
 

python/receive.py

@@ -7,15 +7,15 @@
 channel = connection.channel()
 
 
-channel.queue_declare(queue='test')
+channel.queue_declare(queue='hello')
 
 print ' [*] Waiting for messages. To exit press CTRL+C'
 
 def callback(ch, method, header, body):
     print " [x] Received %r" % (body,)
 
 channel.basic_consume(callback,
-                      queue='test',
+                      queue='hello',
                       no_ack=True)
 
 pika.asyncore_loop()

python/rpc_client.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python
+import pika
+import uuid
+
+class FibonacciClient(object):
+    def __init__(self):
+        self.connection = pika.AsyncoreConnection(pika.ConnectionParameters(
+                host='127.0.0.1',
+                credentials=pika.PlainCredentials('guest', 'guest')))
+        self.channel = self.connection.channel()
+
+        result = self.channel.queue_declare(auto_delete=True)
+        self.callback_queue = result.queue
+
+        self.requests = {}
+        self.channel.basic_consume(self.on_response, no_ack=True,
+                                   queue=self.callback_queue)
+
+    def on_response(self, ch, method, props, body):
+        corr_id = props.correlation_id
+        if corr_id in self.requests:
+            self.requests[corr_id] = body
+
+    def call(self, n):
+        corr_id = str(uuid.uuid4())
+        self.requests[corr_id] = None
+        self.channel.basic_publish(exchange='',
+                                   routing_key='rpc_queue',
+                                   properties=pika.BasicProperties(
+                                         reply_to = self.callback_queue,
+                                         correlation_id = corr_id,
+                                         ),
+                                   body=str(n))
+        while self.requests[corr_id] is None:
+            pika.asyncore_loop(count=1)
+        response = self.requests[corr_id]
+        del self.requests[corr_id]
+        return int(response)
+
+
+fibonacci_rpc = FibonacciClient()
+
+print " [x] Requesting fib(30)"
+response = fibonacci_rpc.call(30)
+print " [.] Got %r" % (response,)
+

python/rpc_server.py

@@ -0,0 +1,39 @@
+#!/usr/bin/env python
+import pika
+
+
+connection = pika.AsyncoreConnection(pika.ConnectionParameters(
+        host='127.0.0.1',
+        credentials=pika.PlainCredentials('guest', 'guest')))
+channel = connection.channel()
+
+
+channel.queue_declare(queue='rpc_queue')
+
+
+def fib(n):
+    if n == 0:
+        return 0
+    elif n == 1:
+        return 1
+    else:
+        return fib(n-1) + fib(n-2)
+
+def on_request(ch, method, props, body):
+    n = int(body)
+
+    print " [.] fib(%s)"  % (n,)
+    response = fib(n)
+
+    ch.basic_publish(exchange='',
+                     routing_key=props.reply_to,
+                     properties=pika.BasicProperties(correlation_id = \
+                                                         props.correlation_id),
+                     body=str(response))
+    ch.basic_ack(delivery_tag = method.delivery_tag)
+
+channel.basic_qos(prefetch_count=1)
+channel.basic_consume(on_request, queue='rpc_queue')
+
+print " [x] Awaiting RPC requests"
+pika.asyncore_loop()

python/send.py

@@ -7,10 +7,10 @@
 channel = connection.channel()
 
 
-channel.queue_declare(queue='test')
+channel.queue_declare(queue='hello')
 
 channel.basic_publish(exchange='',
-                      routing_key='test',
+                      routing_key='hello',
                       body='Hello World!')
 print " [x] Sent 'Hello World!'"
 connection.close()