Database.CQRS.StreamFamily

Documentation

class StreamFamily f fam where

A stream family is a collection of streams of the same type.

For example, a table in a relational database can contain events for all aggregates of a certain type. Each aggregate has its own stream of events but they are all stored in the same table. That table is a stream family indexed by aggregate ID.

Associated Types

type StreamType fam :: *

Type of the streams contained in this stream family.

type StreamIdentifier fam :: *

Identifier for a specific stream, e.g. an aggregate ID.

Methods

getStream :: fam -> StreamIdentifier fam -> f (StreamType fam)

Get the stream corresponding to a given identifier.

allNewEvents :: fam -> f (Producer [(StreamIdentifier fam, Either (EventIdentifier (StreamType fam), String) (EventWithContext' (StreamType fam)))] f a)

Initialise and return a producer of newly-created events from *all* streams in arbitrary batches. If an event can't be decoded, the decoding error is returned instead.

Events should appear in the correct order within a given stream but not necessarily in-between them, i.e. two events belonging to different streams won't necessarily be ordered according to the chronological history.

It is okay for events to be sent more than one time as long as the order is respected within each stream if it makes the implementation easier and prevents the loss of some events.

How events are batched together is up to the implementation as long as the order is respected.

It is okay for batches to be empty to signal that there are currently no new notifications. This is important for migrations, so they know they have processed all events.

latestEventIdentifiers :: fam -> Producer (StreamIdentifier fam, EventIdentifier (StreamType fam)) f ()

Stream the identifier of the latest events for each stream in the family.

It is a snapshot of the last event identifiers at the time the producer is called. It is meant to be used by projections to catch up to the latest state.

Instances

(Eq identifier, Hashable identifier, MonadIO m) => StreamFamily m (StreamFamily identifier metadata event)

Defined in Database.CQRS.InMemory

Associated Types

type StreamType (StreamFamily identifier metadata event) :: Type

type StreamIdentifier (StreamFamily identifier metadata event) :: Type

Methods

getStream :: StreamFamily identifier metadata event -> StreamIdentifier (StreamFamily identifier metadata event) -> m (StreamType (StreamFamily identifier metadata event))

allNewEvents :: StreamFamily identifier metadata event -> m (Producer [(StreamIdentifier (StreamFamily identifier metadata event), Either (EventIdentifier (StreamType (StreamFamily identifier metadata event)), String) (EventWithContext' (StreamType (StreamFamily identifier metadata event))))] m a)

latestEventIdentifiers :: StreamFamily identifier metadata event -> Producer (StreamIdentifier (StreamFamily identifier metadata event), EventIdentifier (StreamType (StreamFamily identifier metadata event))) m ()

Monad m => StreamFamily m (TransformedStreamFamily m streamId eventId metadata event)

Defined in Database.CQRS.Transformer

Associated Types

type StreamType (TransformedStreamFamily m streamId eventId metadata event) :: Type

type StreamIdentifier (TransformedStreamFamily m streamId eventId metadata event) :: Type

Methods

getStream :: TransformedStreamFamily m streamId eventId metadata event -> StreamIdentifier (TransformedStreamFamily m streamId eventId metadata event) -> m (StreamType (TransformedStreamFamily m streamId eventId metadata event))

allNewEvents :: TransformedStreamFamily m streamId eventId metadata event -> m (Producer [(StreamIdentifier (TransformedStreamFamily m streamId eventId metadata event), Either (EventIdentifier (StreamType (TransformedStreamFamily m streamId eventId metadata event)), String) (EventWithContext' (StreamType (TransformedStreamFamily m streamId eventId metadata event))))] m a)

latestEventIdentifiers :: TransformedStreamFamily m streamId eventId metadata event -> Producer (StreamIdentifier (TransformedStreamFamily m streamId eventId metadata event), EventIdentifier (StreamType (TransformedStreamFamily m streamId eventId metadata event))) m ()