How to Stream Join Windowed Mock Topics in Python

import itertools
import random
import time
from collections import deque
from dataclasses import dataclass, field

@dataclass
class Event:
    key: str
    value: int
    timestamp: float = field(default_factory=time.time)

def generate_topic(prefix, keys, start_time):
    while True:
        yield Event(
            key=random.choice(keys),
            value=random.randint(1, 100),
            timestamp=start_time + random.random() * 10
        )

def stream_join(topic_a, topic_b, window_seconds=5.0, max_iterations=20):
    """Join events from two streams when their timestamps fall within a window."""
    a_buffer = deque()
    b_buffer = deque()

    for i in range(max_iterations):
        if i % 2 == 0:
            a_buffer.append(next(topic_a))
        else:
            b_buffer.append(next(topic_b))

        # Remove events outside window from buffers
        current_time = time.time()
        while a_buffer and a_buffer[0].timestamp < current_time - window_seconds:
            a_buffer.popleft()
        while b_buffer and b_buffer[0].timestamp < current_time - window_seconds:
            b_buffer.popleft()

        # Try to match events within window
        for a_event in list(a_buffer):
            for b_event in list(b_buffer):
                if abs(a_event.timestamp - b_event.timestamp) <= window_seconds:
                    print(f"JOIN: A({a_event.key},{a_event.value}) + B({b_event.key},{b_event.value}) "
                          f"delta={abs(a_event.timestamp - b_event.timestamp):.2f}s")
                    a_buffer.remove(a_event)
                    b_buffer.remove(b_event)
                    break

if __name__ == "__main__":
    random.seed(42)
    start = time.time()
    stream_a = generate_topic("A", ["user1", "user2", "user3"], start)
    stream_b = generate_topic("B", ["user2", "user4", "user5"], start)
    stream_join(stream_a, stream_b, window_seconds=5.0)