Publishing Events

Observing the Database#

If the other system can read EventSourcingDB, let it observe the events itself. It then gets every event in the order it was stored, and after a restart, it resumes from the last event it has seen. Nothing needs to be forwarded (see Observing Events).

Publishing with a Projection#

If it can not, because it is a mail server, a webhook, or a message broker, or if it should not depend on how the events of your application look, forward the events with a projection that publishes them instead of writing a view. Do not publish from where a command is executed: if the application stops right after execute, the event is stored, but never published, and nothing tells.

A projection that publishes needs to be resumable (see Resuming Projections), and it reports a failure of the other system as an error of the category Category::Transient. Its events are an enum of their own, which lists only the event type it publishes, so that it is handed nothing else:

use std::{error::Error, fmt};

use architecturekit::{
  Categorized, Category, Envelope, Events, Projection, ResumableProjection, Revision,
};

#[derive(Events)]
pub enum LoanEvent {
  Borrowed(BookBorrowed),
}

pub struct LoanMailer {
  mailer: Mailer,
  checkpoints: CheckpointStore,
}

impl Projection for LoanMailer {
  type Event = LoanEvent;
  type Error = LoanMailerError;

  async fn apply(&self, event: Envelope<LoanEvent>) -> Result<(), LoanMailerError> {
    let LoanEvent::Borrowed(borrowed) = &event.data;

    // The ID of the event lets the receiver recognize a mail it got before.
    self
      .mailer
      .send(event.id, &borrowed.borrowed_by, "You have borrowed a book.")
      .await
      .map_err(LoanMailerError::Mail)
  }
}

impl ResumableProjection for LoanMailer {
  async fn checkpoint(&self) -> Result<Option<Revision>, LoanMailerError> {
    self.checkpoints.load("loan-mailer").await.map_err(LoanMailerError::Checkpoint)
  }

  async fn save_checkpoint(&self, event_id: Revision) -> Result<(), LoanMailerError> {
    self.checkpoints.save("loan-mailer", event_id).await.map_err(LoanMailerError::Checkpoint)
  }
}

#[derive(Debug)]
pub enum LoanMailerError {
  Mail(MailError),
  Checkpoint(CheckpointError),
}

impl fmt::Display for LoanMailerError {
  fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
    match self {
      Self::Mail(error) => write!(formatter, "sending a mail: {error}"),
      Self::Checkpoint(error) => write!(formatter, "keeping the checkpoint: {error}"),
    }
  }
}

impl Error for LoanMailerError {}

impl Categorized for LoanMailerError {
  fn category(&self) -> Option<Category> {
    match self {
      // The mail server may answer again later.
      Self::Mail(_) => Some(Category::Transient),
      Self::Checkpoint(error) => error.category(),
    }
  }
}

Here, Mailer and CheckpointStore stand for whatever your application uses to send mails and to keep a value, and MailError and CheckpointError for the errors they fail with. A projection has one error type for apply, checkpoint, and save_checkpoint, so LoanMailerError covers both. It hands on the category of the errors of the checkpoint store as it is, so the store reports a failure that may pass, such as a lost connection, as an error of the category Category::Transient itself. Otherwise, such a failure ends the run (see Resuming Projections). Start the projection with start_resumable_projection, which goes on from its checkpoint (see Running Projections):

use std::sync::Arc;

use architecturekit::{ProjectionOptions, Subject, Subjects};

let run = store.start_resumable_projection(
  Subjects::tree(Subject::new("/books")),
  Arc::new(LoanMailer { mailer, checkpoints }),
  token.clone(),
  ProjectionOptions::named("loan-mailer"),
);

How a Publisher Behaves#

Since publishing rides on a projection, it behaves like one:

• It resumes only with a checkpoint. A projection started with start_projection starts from the first event whenever the application starts, which suits a view held in memory. A publisher started that way would send every event again on every start. Within a run, the position is kept either way, so reconnecting repeats nothing.
• An event may be published twice. The checkpoint is saved after an event has been applied, so an event published right before the application stopped is published again after the restart. Hand over the ID of the event, so that the receiver can recognize an event it got before.
• A transient failure is tried again. For an error of the category Category::Transient, the run tries again from its checkpoint, with a growing delay. Any other error ends the run, and so does a panic.
• Events arrive in order, one at a time, as they were stored.

On its first start, the checkpoint is empty, so the projection reads every event that has ever been stored, and publishes all of them, for example by sending a mail for every book that has ever been borrowed. To publish only the events from now on, save the ID of the latest event as the checkpoint before the first start. To find it, read the subjects of the projection newest first, and stop after the first event:

use std::pin::pin;

use architecturekit::{Order, ReadOptions, Subject, Subjects};
use futures::StreamExt;

if checkpoints.load("loan-mailer").await?.is_none() {
  let subjects = Subjects::tree(Subject::new("/books"));
  let options = ReadOptions { order: Order::NewestFirst, ..ReadOptions::default() };
  let mut events = pin!(store.read(&subjects, &options).await?);

  if let Some(event) = events.next().await {
    checkpoints.save("loan-mailer", event?.id().parse()?).await?;
  }
}

If there are no events yet, the checkpoint stays empty, since there is nothing to skip.

A transient failure is tried again without a limit. So if the mail server stays down, the run keeps trying, and since it does not end, the liveness probe keeps answering 200, and so does the readiness probe once the run has caught up (see Checking Health over HTTP). To notice it, check the status of the run, whose phase stays Phase::Reconnecting, with since telling since when and error telling why, or log every attempt with the observer of reconnects, whose attempt keeps growing (see Running Projections).

Note that how a projection is run depends on the method that starts it, not on the traits it implements. start_resumable_projection accepts only a projection that implements ResumableProjection, so the compiler makes sure that a publisher started with it keeps a checkpoint.