<commit_before>===========
Projections
===========

This section describes projecting domain events.

Projections can be updated synchronously by direct event subscription, or
asynchronously by following notification logs. Notifications can be pulled,
and pulling can be driven by subscribing to events.

.. contents:: :local:


Subscribing to events
---------------------

The library function
:func:`~eventsourcing.domain.model.decorators.subscribe_to`
can be used to decorate functions, so that the function is called
each time a matching event is published by the library's pub-sub mechanism.

The example below, which is incomplete because the ``TodoView`` is not
defined, suggests that record ``TodoView`` can be created whenever a
``Todo.Created`` event is published. Perhaps there's a ``TodoView`` table
with an index of todo titles, or perhaps it can be joined with a table of users.

.. code:: python

    from eventsourcing.domain.model.decorators import subscribe_to
    from eventsourcing.domain.model.entity import DomainEntity


    class Todo(DomainEntity):
        class Created(DomainEntity.Created):
            pass

    @subscribe_to(Todo.Created)
    def new_todo_projection(event):
        todo = TodoView(id=event.originator_id, title=event.title, user_id=event.user_id)
        todo.save()

It is possible to access aggregates and other views when
updating a view, especially to avoid bloating events with redundant
information that might be added to avoid such queries. Transactions
can be used to insert related records, building a model for a view.

It is possible to use the decorator in a downstream application which
republishes domain events perhaps published by an original application
that uses messaging infrastructure such as an AMQP system. The decorated
function would be called synchronously with the republishing of the event,
but asynchronously with respect to the original application.

A naive projection might consume events as they are published
and update the projection without considering whether the event
was a duplicate, or if previous events were missed.
The trouble with this approach is that, without further modification, without
referring to a fixed sequence and maintaining position in that sequence, there
is no way to recover when events have been missed. If the projection somehow
fails to update when an event is received, then the event will be lost forever to
the projection, and the projection will be forever inconsistent.

Of course, it is possible to follow a fixed sequence of events, for example
using notification logs.


Reading notification logs
-------------------------

If the events of an application are presented as a sequence of
notifications, then the events can be projected using a notification
reader to pull unseen items.

Getting new items can be triggered by pushing prompts to e.g. an AMQP
messaging system, and having the remote components handle the prompts
by pulling the new notifications. To minimise load on the messaging
infrastructure, it may be sufficient simply to send an empty message,
and thereby prompt receivers into pulling new notifications. This may
reduce latency or avoid excessive polling for updates, but the benefit
would be obtained at the cost of additional complexity. Prompts
that include the position of the notification in its sequence would
allow a follower to know when it is being prompted about notifications
it already pulled, and could then skip the pulling operation.

Examples
--------

Application state replication
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Using event record notifications, the state of an application can be
replicated perfectly. If an application can present its event records
as a notification log, then a "replicator" can read the notification
log and write copies of the original records into a replica's record
manager. (If the original application could be partitioned, with each
partition having its own notification log, then the partitions could
be replicated concurrently, which would allow scaling by application
partition. Partitioning an application isn't currently supported in
the library).

In the example below, the ``SimpleApplication`` class is used, which
has a ``RecordManagerNotificationLog`` as its ``notification_log``.
Reading this log, locally or remotely, will yield all the event records
persisted by the ``SimpleApplication``. The ``SimpleApplication``
uses a record manager with contiguous record IDs which allows it to
be used within a record manager notification log object.

A record manager notification log object represents records as record
notifications. With record notifications, the ID of the record in the
notification is used to place the notification in its sequence.
Therefore the ID of the last replicated record is used to determine
the current position in the original application's notification log,
which gives "exactly once" processing.

.. code:: python

    from eventsourcing.application.simple import SimpleApplication
    from eventsourcing.exceptions import ConcurrencyError
    from eventsourcing.domain.model.aggregate import AggregateRoot<commit_msg>Improved projections doc.<commit_after>===========
Projections
===========

This section describes projecting domain events.

Projections can be updated synchronously by direct event subscription, or
asynchronously by following notification logs. Notifications can be pulled,
and pulling can be driven by subscribing to events.

.. contents:: :local:


Subscribing to events
---------------------

The library function
:func:`~eventsourcing.domain.model.decorators.subscribe_to`
can be used to decorate functions, so that the function is called
each time a matching event is published by the library's pub-sub mechanism.

The example below, which is incomplete because the ``TodoView`` is not
defined, suggests that record ``TodoView`` can be created whenever a
``Todo.Created`` event is published. Perhaps there's a ``TodoView`` table
with an index of todo titles, or perhaps it can be joined with a table of users.

.. code:: python

    from eventsourcing.domain.model.decorators import subscribe_to
    from eventsourcing.domain.model.entity import DomainEntity


    class Todo(DomainEntity):
        class Created(DomainEntity.Created):
            pass

    @subscribe_to(Todo.Created)
    def new_todo_projection(event):
        todo = TodoView(id=event.originator_id, title=event.title, user_id=event.user_id)
        todo.save()

It is possible to access aggregates and other views when
updating a view, especially to avoid bloating events with redundant
information that might be added to avoid such queries. Transactions
can be used to insert related records, building a model for a view.

It is possible to use the decorator in a downstream application which
republishes domain events perhaps published by an original application
that uses messaging infrastructure such as an AMQP system. The decorated
function would be called synchronously with the republishing of the event,
but asynchronously with respect to the original application.

A naive projection might consume events as they are published
and update the projection without considering whether the event
was a duplicate, or if previous events were missed.
The trouble with this approach is that, without further modification, without
referring to a fixed sequence and maintaining position in that sequence, there
is no way to recover when events have been missed. If the projection somehow
fails to update when an event is received, then the event will be lost forever to
the projection, and the projection will be forever inconsistent.

Of course, it is possible to follow a fixed sequence of events, for example
using notification logs.


Reading notification logs
-------------------------

If the events of an application are presented as a sequence of
notifications, then the events can be projected using a notification
reader to pull unseen items.

Getting new items can be triggered by pushing prompts to e.g. an AMQP
messaging system, and having the remote components handle the prompts
by pulling the new notifications. To minimise load on the messaging
infrastructure, it may be sufficient simply to send an empty message,
and thereby prompt receivers into pulling new notifications. This may
reduce latency or avoid excessive polling for updates, but the benefit
would be obtained at the cost of additional complexity. Prompts
that include the position of the notification in its sequence would
allow a follower to know when it is being prompted about notifications
it already pulled, and could then skip the pulling operation.

Examples
--------

Application state replication
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Using event record notifications, the state of an application can be
replicated perfectly. If an application can present its event records
as a notification log, then a "replicator" can read the notification
log and write copies of the original records into a replica's record
manager. (If the original application could be partitioned, with each
partition having its own notification log, then the partitions could
be replicated concurrently, which would allow scaling by application
partition. Partitioning an application isn't currently supported in
the library.)

In the example below, the ``SimpleApplication`` class is used, which
has a ``RecordManagerNotificationLog`` as its ``notification_log``.
Reading this log, locally or remotely, will yield all the event records
persisted by the ``SimpleApplication``. The ``SimpleApplication``
uses a record manager with contiguous record IDs which allows it to
be used within a record manager notification log object.

A record manager notification log object represents records as record
notifications. With record notifications, the ID of the record in the
notification is used to place the notification in its sequence.
Therefore the ID of the last replicated record is used to determine
the current position in the original application's notification log,
which gives "exactly once" processing.

.. code:: python

    from eventsourcing.application.simple import SimpleApplication
    from eventsourcing.exceptions import ConcurrencyError
    from eventsourcing.domain.model.aggregate import AggregateRoot
{'fraction_non_alphanumeric': 0.055447659739755295, 'fraction_numerical': 0.0, 'mean_word_length': 4.646381578947368, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3279495', 'n_tokens_mistral': 2362, 'n_tokens_neox': 2219, 'n_words': 1385}