Integrating Eventsourcing with a Message Broker

Hello! Welcome back to our course on distributed systems architecture.

In our previous lesson, we compared a manual event sourcing implementation with the eventsourcing library. We concluded that for non-trivial systems, the library's abstractions for persistence, snapshotting, and notifications provide significant advantages in productivity and long-term maintainability.

Today, we will build directly on that foundation by focusing on one of the most critical features for distributed systems: event notification. This lesson addresses the learning outcome:

Set up the 'eventsourcing' library with a message broker integration for event notification.

We will explore the library's core mechanism for propagating events—the NotificationLog—and see how it enables communication between different parts of a system. This is the fundamental pattern for building event-driven architectures, where services can react to state changes in others without being tightly coupled. We will first understand the library's internal communication model and then bridge that concept to integrating with an external message broker like RabbitMQ or Kafka.

1. The NotificationLog: The Application's Public Record

At the heart of the library's notification system is the NotificationLog. Conceptually, it is a single, append-only, totally-ordered log of every event that occurs across all aggregates within an application. This log serves as the definitive source of truth for any downstream system that needs to react to state changes.

The Application class, which we explored previously, automatically maintains this log. Every time you call application.save(), the new events are atomically persisted to their aggregate's stream and recorded in the application's notification log.

To understand the mechanics, let's look at the documentation.

As you read, note these key points:

• The Application class has a log attribute, which is an instance of LocalNotificationLog.
• This log contains Notification objects, each with a unique, sequential integer ID.
• The log is read in "sections" (e.g., application.log["1,10"]), which is a design pattern for efficiently pulling batches of events over a network.

The performance and scalability of this log are critical. The library uses a clever data structure called BigArray to ensure that appending and reading from the log remain efficient even as it grows to contain billions of events.

This design, which uses a globally ordered sequence of events, is fundamental. It provides a reliable mechanism for other components, or "projectors," to consume the event stream without missing events. A consumer simply needs to track the ID of the last event it processed and ask the log for all subsequent events.

2. The System Pattern: In-Process Event-Driven Communication

Before integrating an external message broker, it's essential to understand the library's native model for inter-service communication. The eventsourcing library provides a System class that wires together multiple Application objects, allowing them to communicate via their notification logs.

This pattern introduces two key roles:

• Leader: An application that produces events and publishes notifications.
• Follower: An application that listens for notifications, pulls events from a Leader's notification log, and processes them. A Follower implements a policy() method that decides how to react to each incoming event.

An application that both consumes and produces events is often called a ProcessApplication.

This Leader/Follower setup forms a processing pipeline. Let's examine how this is implemented.

The System class, combined with a Runner (like SingleThreadedRunner), effectively creates an in-memory message bus.

1. The WorldsApplication saves a World.SomethingHappened event.
2. The Runner detects this and "prompts" the Counters application.
3. The Counters application pulls the new event from the WorldsApplication's notification log.
4. It processes the event in its policy() method, finds or creates a Counter aggregate, and increments it.

This architecture allows you to build complex, decoupled systems where one service (e.g., Counters) builds its state entirely by reacting to events published by another (WorldsApplication).

3. Bridging to an External Message Broker

The System class is perfect for applications running within the same process or on the same machine. However, in a true distributed system, services run independently and communicate over a network. This is where a message broker like RabbitMQ or Kafka comes in.

The eventsourcing library does not provide a direct, built-in adapter for a specific broker. Instead, it provides the essential primitive—the NotificationLog—and you build the integration around it. This is a deliberate design choice that keeps the library decoupled from any specific messaging technology.

The process involves two main components: a Publisher and a Subscriber.

The Publisher: Reading the Log and Sending to the Broker

The publisher is a dedicated process whose job is to tail the NotificationLog of your primary application and publish each new event notification to the message broker.

This publisher would look something like this in pseudocode:

import time
from eventsourcing.application import Application
from eventsourcing.interface.notificationlog import NotificationLogReader
# Assume 'message_broker_client' is a pre-configured client for RabbitMQ, Kafka, etc.

def run_publisher(application: Application):
    # Get the application's notification log
    notification_log = application.log

    # Create a reader to poll the log
    reader = NotificationLogReader(notification_log)

    # Persistently store the ID of the last processed notification
    # (e.g., in a file, database, or Redis)
    last_seen_id = load_last_seen_id() 

    while True:
        # Read all new notifications since the last one we saw
        notifications = reader.read(start=last_seen_id + 1)
        
        new_notifications = list(notifications)
        if new_notifications:
            for notification in new_notifications:
                # Publish the notification to the broker
                # The notification object can be serialized (e.g., to JSON)
                message_broker_client.publish(
                    topic='domain_events', 
                    message=notification.serialize()
                )
                
                # Update the last seen ID
                last_seen_id = notification.id
                
            # Persist the latest processed ID
            save_last_seen_id(last_seen_id)
        else:
            # If no new events, wait a bit before polling again
            time.sleep(1)

This "pull-then-push" model is highly resilient. If the publisher process crashes, it can restart, read the last successfully published ID, and resume from that exact point in the NotificationLog, ensuring no events are lost.

The Subscriber: Consuming from the Broker

The subscriber is a standard consumer for your chosen message broker. It listens on the appropriate topic or queue, receives the serialized event notification, and processes it. This processing could involve:

• Updating a denormalized read model (for CQRS).
• Triggering a command in another bounded context.
• Calling an external API.

This architecture cleanly separates the event-producing application from the event-consuming services, which is a hallmark of robust distributed systems.

Conclusion

In this lesson, we have demystified how the eventsourcing library enables the creation of event-driven systems. We've seen that it provides the fundamental building blocks for reliable event propagation, which can be used both for in-process communication and for integration with external, network-based message brokers.

Key Takeaways:

• The NotificationLog is the library's core primitive for event propagation. It is a scalable, totally-ordered, append-only log of all events in an application.
• The System class with its Leader/Follower pattern provides a powerful way to compose event-driven applications that communicate in-process. This is ideal for building projectors and other coupled services.
• To integrate with an external message broker, you create a separate publisher process. This process uses a NotificationLogReader to poll for new events and publishes them to the broker, bridging the eventsourcing application with the wider distributed system.
• This "pull-from-log, push-to-broker" pattern provides high-fidelity event delivery, as consumers are decoupled and the publisher can reliably resume after failures.

Preview of the Next Lesson:

We've now established the architectural pattern for connecting our event-sourced application to a message broker. In the first lesson of the next module, we will get practical and set up the broker itself. The lesson will be: "Install and run a RabbitMQ instance locally using Docker." This will prepare our environment for implementing the publisher/subscriber patterns we've discussed today.