11.4Pattern 3: Queue Drain

Read from a message queue until all expected events have arrived or timeout.

def drain_queue(queue_url, expected_count, timeout=30):
    """Read messages from a queue until expected count reached or timeout."""
    messages = []
    deadline = time.time() + timeout

    while len(messages) < expected_count and time.time() < deadline:
        batch = poll_sqs(queue_url, timeout=2)
        if batch:
            messages.append(batch)

    return messages

# Usage
def test_order_generates_three_events(api_client, sqs_queues):
    """Creating an order should produce events to billing, shipping, and inventory."""
    api_client.post("/api/orders", json=ORDER_PAYLOAD)

    # Drain all three queues
    billing = drain_queue(sqs_queues["billing"], expected_count=1, timeout=10)
    shipping = drain_queue(sqs_queues["shipping"], expected_count=1, timeout=10)
    inventory = drain_queue(sqs_queues["inventory"], expected_count=1, timeout=10)

    assert len(billing) == 1, "Missing billing event"
    assert len(shipping) == 1, "Missing shipping event"
    assert len(inventory) == 1, "Missing inventory event"

When to use: Event-driven systems where one action produces events to multiple queues.