pika-things.py

#!/usr/bin/env python
import pika
import time
import sys
import json

name = 'hello ' + sys.argv[1]

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))

channel = connection.channel()

channel.queue_declare(queue=name)
channel.queue_bind(queue=name, exchange='pika', routing_key='request')


def callback(ch, method, properties, body):
    body = json.loads(body)
    data = body['data']

    ns = {}
    exec(body['fn'], ns)
    the_script = ns['the_script']
    time.sleep(2)

    result = the_script(name, data)
    channel.basic_publish(exchange='pika',
                          routing_key='results',
                          body=str(result))

    channel.basic_ack(method.delivery_tag)


channel.basic_consume(callback,
                      queue=name,
                      no_ack=False)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

###### ^^ this was the client ####

#!/usr/bin/env python
import pika
from random import randint
import json
import inspect
import time

# setting up the connection
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='pika',
                         exchange_type='direct',
                         durable=False,
                         auto_delete=True)

channel.queue_declare('results')
channel.queue_bind(queue='results',
                   exchange='pika',
                   routing_key='results')

# the method to send


def the_script(name, n):
    res = n * 2
    return "{0} * 2 = {1} [{2}]".format(n, res, name)


def send(num):
    fn_txt = "".join(inspect.getsourcelines(the_script)[0])
    data = dict(fn=fn_txt, data=num)

    channel.basic_publish(exchange='pika',
                          routing_key='request',
                          body=json.dumps(data))
    print(" [x] Sent '%s'" % json.dumps(data['data']))


def recieved(m, body):
    print(body)
    channel.basic_ack(m.delivery_tag)


while True:
    m, _, body = channel.basic_get(queue='results', no_ack=False)
    if m:
        recieved(m, body)
    send(randint(1, 100))
    time.sleep(1)