amqp-trace.py

#!../venv/bin/python

import pika, pprint, json
import time

import logging
logging.basicConfig()

def jsonlog(channel, method_frame, header_frame, body):
    msg = json.loads(body)
    print time.strftime("%c", time.localtime(msg['created'])), method_frame.routing_key
    print msg['pathname']
    print msg['msg']
    print "-=-=-"

def simple(channel, method, header, body):
    print time.strftime("%c", time.localtime())
    print body
    print "-=-=-"

def consume(url, exchange, routing_key, callback=simple):
    state = {'channel': None, 'queue': None}

    def connected(connection):
        print "-- connected"
        connection.channel(channel_open)

    def channel_open(new_channel):
        print "-- channel open"
        state['channel'] = new_channel
        state['channel'].queue_declare(callback=queue_declared, exclusive=True)

    def queue_declared(frame):
        print "-- queue declared:", frame.method.queue
        state['queue'] = frame.method.queue
        state['channel'].queue_bind(callback=queue_bound, exchange=exchange, queue=state['queue'], routing_key=routing_key)

    def queue_bound(frame):
        print "-- queue bound"
        state['channel'].basic_consume(callback, queue=state['queue'], no_ack=True)

    conn = pika.adapters.SelectConnection(pika.connection.URLParameters(url), connected)
    try:
        print "-- starting..."
        conn.ioloop.start()
    except KeyboardInterrupt:
        print ""
        print "-- stopping..."
        conn.close()
        conn.ioloop.start()
        print "-- goodbye!"

if __name__=="__main__":
    import sys

    url = "amqp://guest:[email protected]:5672/%2F"

    if sys.argv[2:]:
        routing_key = sys.argv[2]
    else:
        routing_key = "#"

    if sys.argv[1:]:
        exchange = sys.argv[1]
    else:
        exchange = "logs.production"

    print "Connection:", url
    print "Exchange:", exchange
    print "Route key:", routing_key

    consume(url, exchange, routing_key)