Pub/Sub and Message Brokers

RabbitMQ Consumer

def process_messages(ch, method, properties, body):
    data = json.loads(body)
    result = compiled_transform(data)
    ch.basic_publish(exchange='', routing_key='results', body=json.dumps(result))

Kafka Stream Processing

from kafka import KafkaConsumer

consumer = KafkaConsumer('events')
for msg in consumer:
    result = compiled_process(msg.value)
    producer.send('results', result)

Redis Pub/Sub

import redis
r = redis.Redis()
p = r.pubsub()
p.subscribe('compute-channel')
for message in p.listen():
    if message['type'] == 'message':
        result = compiled_compute(message['data'])