puka: six.

Author: majek

python-puka/README.md

@@ -49,3 +49,8 @@ You may need to install `pip` first:
 
  python receive_logs_topic.py
  python emit_log_topic.py
+
+[Tutorial six: RPC](http://www.rabbitmq.com/tutorial-six-python.html):
+
+ python rpc_server.py
+ python rpc_client.py

python-puka/rpc_client.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+import puka
+import uuid
+
+class FibonacciRpcClient(object):
+ def __init__(self):
+ self.client = client = puka.Client("amqp://localhost/")
+ promise = client.connect()
+ client.wait(promise)
+
+ promise = client.queue_declare(exclusive=True)
+ self.callback_queue = client.wait(promise)['queue']
+
+ self.consume_promise = client.basic_consume(queue=self.callback_queue,
+ no_ack=True)
+
+ def call(self, n):
+ correlation_id = str(uuid.uuid4())
+ # We don't need to wait on promise from publish, let it happen async.
+ self.client.basic_publish(exchange='',
+ routing_key='rpc_queue',
+ headers={'reply_to': self.callback_queue,
+ 'correlation_id': correlation_id},
+ body=str(n))
+ while True:
+ msg_result = self.client.wait(self.consume_promise)
+ if msg_result['headers']['correlation_id'] == correlation_id:
+ return int(msg_result['body'])
+
+
+fibonacci_rpc = FibonacciRpcClient()
+
+print " [x] Requesting fib(30)"
+response = fibonacci_rpc.call(30)
+print " [.] Got %r" % (response,)

python-puka/rpc_server.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python
+import puka
+
+client = puka.Client("amqp://localhost/")
+promise = client.connect()
+client.wait(promise)
+
+promise = client.queue_declare(queue='rpc_queue')
+client.wait(promise)
+
+# The worlds worst algorithm:
+def fib(n):
+ if n == 0:
+ return 0
+ elif n == 1:
+ return 1
+ else:
+ return fib(n-1) + fib(n-2)
+
+
+print " [x] Awaiting RPC requests"
+consume_promise = client.basic_consume(queue='rpc_queue', prefetch_count=1)
+while True:
+ msg_result = client.wait(consume_promise)
+ n = int(msg_result['body'])
+
+ print " [.] fib(%s)" % (n,)
+ response = fib(n)
+
+ # This publish doesn't need to be synchronous.
+ client.basic_publish(exchange='',
+ routing_key=msg_result['headers']['reply_to'],
+ headers={'correlation_id':
+ msg_result['headers']['correlation_id']},
+ body=str(response))
+ client.basic_ack(msg_result)