amqpcat

amqpcat.py

#!/usr/bin/env python3

from yaml import safe_load_all, dump
import sys
import pika

def serialize(channel, method, props, body):
    msg={'body': body.decode('utf-8'), 'headers': props.headers, 'routingkey': method.routing_key}
    dump(msg, sys.stdout, default_flow_style=False)
    sys.stdout.write('---\n')
    sys.stdout.flush()

def reading(channel, q):
    channel.basic_consume(queue=q, no_ack=True, consumer_callback=serialize)
    channel.start_consuming()

def writing(channel, exg):
    msgs = safe_load_all(sys.stdin.buffer.raw)
    for msg in msgs:
        if isinstance(msg, dict) and 'body' in msg:
            props = pika.spec.BasicProperties(headers=msg.get('headers', {}))
            channel.basic_publish(exchange=exg,
                                  routing_key=msg.get('routingkey', ''),
                                  properties=props,
                                  body=msg['body'])

def amqpcat(parameters):
    amqpParam = pika.URLParameters(parameters[1])
    connection = pika.BlockingConnection(amqpParam)
    channel = connection.channel()
    if parameters[0] == 'from':
        reading(channel, parameters[2])
    elif parameters[0] == 'to':
        writing(channel, parameters[2])
    

if __name__ == '__main__':
    amqpcat(sys.argv[1:])