Database Tutorial #15: Redis Pub/Sub and Streams

# Terminal 1: Subscribe to a channel
SUBSCRIBE notifications:user:1

# Terminal 2: Publish
PUBLISH notifications:user:1 '{"type":"message","text":"Hello!"}'
# Terminal 1 receives: "Hello!"

import Redis from "ioredis";

// Publisher client (normal connection)
const publisher = new Redis();

// Subscriber client (dedicated — can only subscribe)
const subscriber = new Redis();

// Subscribe
await subscriber.subscribe("notifications");
subscriber.on("message", (channel: string, message: string) => {
  const data = JSON.parse(message);
  console.log(`[${channel}]`, data);
});

// Publish from anywhere
async function sendNotification(userId: number, text: string) {
  await publisher.publish("notifications", JSON.stringify({
    userId,
    text,
    timestamp: Date.now(),
  }));
}

await subscriber.psubscribe("notifications:*");
subscriber.on("pmessage", (pattern, channel, message) => {
  console.log(pattern, channel, JSON.parse(message));
});

# Add to stream (auto-generates ID: timestamp-sequence)
XADD orders * event "order_placed" order_id "123" amount "99.99"
# Returns: "1726500000000-0"

# Read from stream
XREAD COUNT 10 STREAMS orders 0

# Read new messages only ($ = latest)
XREAD COUNT 10 BLOCK 0 STREAMS orders $

# Create consumer group (starting from the beginning)
XGROUP CREATE orders processing-group 0 MKSTREAM

# Consumer reads from group
XREADGROUP GROUP processing-group worker1 COUNT 1 STREAMS orders >
# > means "give me new, undelivered messages"

# Acknowledge after processing
XACK orders processing-group 1726500000000-0

import Redis from "ioredis";

const redis = new Redis();

// Producer: add events to stream
async function placeOrder(orderId: string, amount: number) {
  const id = await redis.xadd("orders", "*",
    "event", "order_placed",
    "order_id", orderId,
    "amount", String(amount)
  );
  console.log("Event added:", id);
}

// Consumer: process events from consumer group
async function startWorker(workerId: string) {
  const GROUP = "processing-group";
  const STREAM = "orders";

  // Create group if it doesn't exist
  await redis.xgroup("CREATE", STREAM, GROUP, "0", "MKSTREAM").catch(() => {});

  console.log(`Worker ${workerId} started`);

  while (true) {
    // Read up to 1 message, block 2 seconds if empty
    const results = await redis.xreadgroup(
      "GROUP", GROUP, workerId,
      "COUNT", "1",
      "BLOCK", "2000",
      "STREAMS", STREAM, ">"
    ) as any;

    if (!results) continue;

    for (const [stream, messages] of results) {
      for (const [id, fields] of messages) {
        const data = Object.fromEntries(
          fields.reduce((acc: any[], v: string, i: number) =>
            i % 2 === 0 ? [...acc, [v, fields[i + 1]]] : acc, [])
        );

        console.log(`Processing ${id}:`, data);

        try {
          await processOrder(data);
          await redis.xack(STREAM, GROUP, id);
        } catch (err) {
          console.error("Processing failed:", err);
          // Message stays in pending list — will be retried or reassigned
        }
      }
    }
  }
}

async function processOrder(data: Record<string, string>) {
  // Your business logic here
  console.log("Order processed:", data.order_id);
}

import redis
import time
import json

r = redis.Redis(host="localhost", decode_responses=True)

# Producer
def place_order(order_id: str, amount: float):
    entry_id = r.xadd("orders", {
        "event": "order_placed",
        "order_id": order_id,
        "amount": str(amount),
    })
    print(f"Event added: {entry_id}")

# Consumer worker
def start_worker(worker_id: str):
    group = "processing-group"
    stream = "orders"

    # Create group if needed
    try:
        r.xgroup_create(stream, group, id="0", mkstream=True)
    except redis.exceptions.ResponseError:
        pass  # Group already exists

    print(f"Worker {worker_id} started")

    while True:
        # Read up to 1 message, block 2 seconds
        results = r.xreadgroup(group, worker_id, {stream: ">"}, count=1, block=2000)

        if not results:
            continue

        for stream_name, messages in results:
            for entry_id, data in messages:
                print(f"Processing {entry_id}:", data)
                try:
                    # Process the event
                    print(f"Order processed: {data['order_id']}")
                    r.xack(stream_name, group, entry_id)
                except Exception as e:
                    print(f"Error: {e}")