Monads are hard. Oh, the irony!

pika_rpc.py

"""
Or should i use bass-ackwards definitions?.. http://pika.readthedocs.org/en/0.9.6/examples/comparing_publishing_sync_async.html
"""

import pika

pika_conn = 'amqp://guest:guest@localhost:5672/%2F'


def call(uri, service, method, **query):
    state = {'query': json.dumps(query)}

    def connected(conn):
        state['connection'] = conn
        conn.channel(got_channel)

    def got_channel(chan):
        state['chan'] = chan
        state['chan'].exchange_declare(exchange_in_ok, exchange=service, passive=True)

    def exchange_in_ok(_):
        state['chan'].exchange_declare(exchange_out_ok, exchange="result", exchange_type="direct", durable=True)

    def exchange_out_ok(_):
        state['chan'].queue_declare(res_queue_declared, queue="", exclusive=True)

    def res_queue_declared(frame):
        state['queue'] = frame.method.queue
        state['chan'].queue_bind(go_publish, exchange="result", queue=state['queue'], routing_key=state['queue'])

    def go_publish(_):
        props = pika.BasicProperties(content_type="application/json", reply_to=state['queue'])
        state['chan'].basic_publish(exchange=service, routing_key=method, properties=props, body=state['query'])
        state['chan'].basic_consume(got_result, queue=state['queue'], no_ack=True)

    def got_result(_chan, _envelope, _props, body):
        state['result'] = json.loads(body)
        state['connection'].close()

    conn = pika.adapters.SelectConnection(pika.connection.URLParameters(uri), connected)
    try:
        conn.ioloop.start()
    except KeyboardInterrupt:
        conn.close()
        conn.ioloop.start()

    return state['result']