application

The application layer brings together the domain and infrastructure layers.

simple

class eventsourcing.application.simple.SimpleApplication(name='', persistence_policy=None, persist_event_type=None, cipher_key=None, sequenced_item_class=None, sequenced_item_mapper_class=None, record_manager_class=None, stored_event_record_class=None, event_store_class=None, snapshot_record_class=None, setup_table=True, contiguous_record_ids=True, pipeline_id=0, json_encoder_class=None, json_decoder_class=None, notification_log_section_size=None)

Bases: eventsourcing.application.pipeline.Pipeable

Base class for event sourced applications.

Constructs infrastructure objects such as the repository and event store, and also the notification log which presents the application state as a sequence of events.

Needs actual infrastructure classes.

infrastructure_factory_class

alias of eventsourcing.infrastructure.factory.InfrastructureFactory

__init__(name='', persistence_policy=None, persist_event_type=None, cipher_key=None, sequenced_item_class=None, sequenced_item_mapper_class=None, record_manager_class=None, stored_event_record_class=None, event_store_class=None, snapshot_record_class=None, setup_table=True, contiguous_record_ids=True, pipeline_id=0, json_encoder_class=None, json_decoder_class=None, notification_log_section_size=None)

Initialize self. See help(type(self)) for accurate signature.

event_store_class

alias of eventsourcing.infrastructure.eventstore.EventStore

construct_infrastructure_factory(*args, **kwargs)

Return type:InfrastructureFactory

class eventsourcing.application.simple.ApplicationWithConcreteInfrastructure(name='', persistence_policy=None, persist_event_type=None, cipher_key=None, sequenced_item_class=None, sequenced_item_mapper_class=None, record_manager_class=None, stored_event_record_class=None, event_store_class=None, snapshot_record_class=None, setup_table=True, contiguous_record_ids=True, pipeline_id=0, json_encoder_class=None, json_decoder_class=None, notification_log_section_size=None)

Bases: eventsourcing.application.simple.SimpleApplication

Subclasses have actual infrastructure.

snapshotting

class eventsourcing.application.snapshotting.SnapshottingApplication(snapshot_period=None, **kwargs)

Bases: eventsourcing.application.simple.SimpleApplication

__init__(snapshot_period=None, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

policies

class eventsourcing.application.policies.PersistencePolicy(event_store, persist_event_type=None)

Bases: object

Stores events of given type to given event store, whenever they are published.

__init__(event_store, persist_event_type=None)

Initialize self. See help(type(self)) for accurate signature.

notificationlog

class eventsourcing.application.notificationlog.AbstractNotificationLog

Bases: abc.ABC

Presents a sequence of sections from a sequence of notifications.

__getitem__(section_id)

Get section of notification log.

Return type:Section

class eventsourcing.application.notificationlog.Section(section_id, items, previous_id=None, next_id=None)

Bases: object

Section of a notification log.

Contains items, and has an ID.

May also have either IDs of previous and next sections of the notification log.

__init__(section_id, items, previous_id=None, next_id=None)

Initialize self. See help(type(self)) for accurate signature.

class eventsourcing.application.notificationlog.LocalNotificationLog(section_size=None)

Bases: eventsourcing.application.notificationlog.AbstractNotificationLog

Presents a sequence of sections from a sequence of notifications.

__init__(section_size=None)

Initialize self. See help(type(self)) for accurate signature.

__getitem__(section_id)

Get section of notification log.

Return type:Section

get_next_position()

Returns next unoccupied position in zero-based sequence.

Since the notification IDs are one-based, the next position in the zero-based sequence equals the current max notification ID which is 1-based. If there are no records, the max notification ID will be None, and the next position is zero.

Returns:Non-negative integer. Return type:int

get_items(start, stop, next_position=None)

Returns items for section.

Return type:list

class eventsourcing.application.notificationlog.RecordManagerNotificationLog(record_manager, section_size)

Bases: eventsourcing.application.notificationlog.LocalNotificationLog

__init__(record_manager, section_size)

Initialize self. See help(type(self)) for accurate signature.

get_items(start, stop, next_position=None)

Returns items for section.

Return type:list

get_next_position()

Returns next unoccupied position in zero-based sequence.

Since the notification IDs are one-based, the next position in the zero-based sequence equals the current max notification ID which is 1-based. If there are no records, the max notification ID will be None, and the next position is zero.

Returns:Non-negative integer. Return type:int

class eventsourcing.application.notificationlog.BigArrayNotificationLog(big_array, section_size)

Bases: eventsourcing.application.notificationlog.LocalNotificationLog

__init__(big_array, section_size)

Initialize self. See help(type(self)) for accurate signature.

get_items(start, stop, next_position=None)

Returns items for section.

Return type:list

get_next_position()

Returns next unoccupied position in zero-based sequence.

Returns:Non-negative integer. Return type:int

class eventsourcing.application.notificationlog.NotificationLogReader(notification_log, use_direct_query_if_available=False)

Bases: abc.ABC

__init__(notification_log, use_direct_query_if_available=False)

Initialize self. See help(type(self)) for accurate signature.

seek(position)

Sets position of reader in notification log sequence.

This represents the position of the last notification read by the reader. The next notification returned by the reader will be the next position.

Parameters:position (int) – Position is notification log sequence. Raises:ValueError – if the position is less than zero

initial_section_id

Returns initial section ID used to start getting linked sections from the notification log.

Slight departure from Vaughn Vernon’s design by not using ‘current’ as initial section ID, but a section ID that just includes the “next” position, which the notification log can use to return the section containing this position. This avoids lengthy back- tracking when reader has a lot of notifications to catch-up on.

This property has been extracted in order to allow a subclass to adjust this default behaviour.

It would be possible to calculate the actual section ID from the current reader position. Using section ID of an actual section may hit a cache and avoid troubling the server, but the reader would need to know the section size of the notification log it is reading. If we don’t know section size, perhaps it is a remote notification log, we can use ‘current’ to hit a cache. In future, it might be possible to ask the notification log to disclose it’s section size, or compute an actual section ID for a given position.

Returns:A notification log section ID. Return type:str

django

class eventsourcing.application.django.DjangoApplication(tracking_record_class=None, *args, **kwargs)

Bases: eventsourcing.application.simple.ApplicationWithConcreteInfrastructure

infrastructure_factory_class

alias of eventsourcing.infrastructure.django.factory.DjangoInfrastructureFactory

__init__(tracking_record_class=None, *args, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

classmethod reset_connection_after_forking()

Resets database connection after forking.

sqlalchemy

class eventsourcing.application.sqlalchemy.SQLAlchemyApplication(uri=None, session=None, tracking_record_class=None, **kwargs)

Bases: eventsourcing.application.simple.ApplicationWithConcreteInfrastructure

infrastructure_factory_class

alias of eventsourcing.infrastructure.sqlalchemy.factory.SQLAlchemyInfrastructureFactory

stored_event_record_class

alias of eventsourcing.infrastructure.sqlalchemy.records.StoredEventRecord

snapshot_record_class

alias of eventsourcing.infrastructure.sqlalchemy.records.EntitySnapshotRecord

__init__(uri=None, session=None, tracking_record_class=None, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

popo

class eventsourcing.application.popo.PopoApplication(name='', persistence_policy=None, persist_event_type=None, cipher_key=None, sequenced_item_class=None, sequenced_item_mapper_class=None, record_manager_class=None, stored_event_record_class=None, event_store_class=None, snapshot_record_class=None, setup_table=True, contiguous_record_ids=True, pipeline_id=0, json_encoder_class=None, json_decoder_class=None, notification_log_section_size=None)

Bases: eventsourcing.application.simple.ApplicationWithConcreteInfrastructure

infrastructure_factory_class

alias of eventsourcing.infrastructure.popo.factory.PopoInfrastructureFactory

sequenced_item_mapper_class

alias of eventsourcing.infrastructure.popo.mapper.SequencedItemMapperForPopo

pipeline

class eventsourcing.application.pipeline.PipelineExpression(left, right)

Bases: object

Implements a left-to-right association between two objects.

__init__(left, right)

Initialize self. See help(type(self)) for accurate signature.

class eventsourcing.application.pipeline.PipeableMetaclass

Bases: abc.ABCMeta

Meta class for pipeable classes.

__or__(other)

Implements bitwise or operator ‘|’ as a pipe between pipeable classes.

class eventsourcing.application.pipeline.Pipeable

Bases: object

Base class for pipeable classes.

process

class eventsourcing.application.process.ProcessApplication(name=None, policy=None, setup_table=False, use_direct_query_if_available=False, notification_log_reader_class=None, **kwargs)

Bases: eventsourcing.application.simple.SimpleApplication

__init__(name=None, policy=None, setup_table=False, use_direct_query_if_available=False, notification_log_reader_class=None, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

notification_log_reader_class

alias of eventsourcing.application.notificationlog.NotificationLogReader

publish_prompt(event=None)

Publishes prompt for a given event.

Used to prompt downstream process application when an event is published by this application’s model, which can happen when application command methods, rather than the process policy, are called.

Wraps exceptions with PromptFailed, to avoid application policy exceptions being seen directly in other applications when running synchronously in single thread.

static policy(repository, event)

Empty method, can be overridden in subclasses to implement concrete policy.

class eventsourcing.application.process.WrappedRepository(repository: eventsourcing.infrastructure.eventsourcedrepository.EventSourcedRepository)

Bases: object

Used to wrap an event sourced repository for use in process application policy so that use of, and changes to, domain model aggregates can be automatically detected and recorded.

Implements a “dictionary like” interface, so that aggregates can be accessed by ID.

__init__(repository: eventsourcing.infrastructure.eventsourcedrepository.EventSourcedRepository)

Initialize self. See help(type(self)) for accurate signature.

save_orm_obj(orm_obj)

Includes orm_obj in “process event”, so that projections into custom ORM objects is as reliable with respect to sudden restarts as “normal” domain event processing in a process application.

class eventsourcing.application.process.ProcessApplicationWithSnapshotting(snapshot_period=None, **kwargs)

Bases: eventsourcing.application.snapshotting.SnapshottingApplication, eventsourcing.application.process.ProcessApplication

command

class eventsourcing.application.command.CommandProcess(name=None, policy=None, setup_table=False, use_direct_query_if_available=False, notification_log_reader_class=None, **kwargs)

Bases: eventsourcing.application.process.ProcessApplication

persist_event_type

alias of eventsourcing.domain.model.command.Command.Event

system

class eventsourcing.application.system.System(*pipeline_exprs, **kwargs)

Bases: object

A system object has a set of pipeline expressions, which involve process application classes. A system object can be run using a system runner.

__init__(*pipeline_exprs, **kwargs)

Initialises a “process network” system object.

Parameters:pipeline_exprs – Pipeline expressions involving process application classes.

Each pipeline expression of process classes shows directly which process follows which other process in the system.

For example, the pipeline expression (A | B | C) shows that B follows A, and C follows B.

The pipeline expression (A | A) shows that A follows A.

The pipeline expression (A | B | A) shows that B follows A, and A follows B.

The pipeline expressions ((A | B | A), (A | C | A)) are equivalent to (A | B | A | C | A).

construct_app(process_class, infrastructure_class=None, **kwargs)

Constructs process application from given process_class.

is_prompt(event)

Predicate for subscribing to publication of Prompt objects.

__enter__()

Supports usage of a system object as a context manager.

__exit__(exc_type, exc_val, exc_tb)

Supports usage of a system object as a context manager.

__getattr__(process_name, infrastructure_class=None)

Supports accessing process application by name as object attribute.

Parameters:

• process_name –
• infrastructure_class –

Returns:

A process application object.

Return type:

ProcessApplication

bind(infrastructure_class)

Constructs a system object that has an infrastructure class from system object constructed without infrastructure class.

Raises ProgrammingError if already have an infrastructure class.

Parameters:infrastructure_class – Returns:System object that has an infrastructure class. Return type:System

class eventsourcing.application.system.SystemRunner(system: eventsourcing.application.system.System, infrastructure_class=None, setup_tables=False, use_direct_query_if_available=False)

Bases: abc.ABC

__init__(system: eventsourcing.application.system.System, infrastructure_class=None, setup_tables=False, use_direct_query_if_available=False)

Initialize self. See help(type(self)) for accurate signature.

__enter__()

Supports usage of a system runner as a context manager.

__exit__(exc_type, exc_val, exc_tb)

Supports usage of a system runner as a context manager.

start()

Starts running the system.

Abstract method which must be implemented on concrete descendants.

close()

Closes a running system.

__getattr__(process_name)

Supports accessing process application by name as object attribute.

Parameters:

• process_name –
• infrastructure_class –

Returns:

A process application object.

Return type:

ProcessApplication

class eventsourcing.application.system.InProcessRunner(system: eventsourcing.application.system.System, infrastructure_class=None, setup_tables=False, use_direct_query_if_available=False)

Bases: eventsourcing.application.system.SystemRunner

Runs a system in the current process, either in the current thread, or with one thread for each process in the system.

start()

Starts running the system.

Abstract method which must be implemented on concrete descendants.

handle_prompt(prompt)

Handles publication of a prompt.

Abstract method which must be implemented on concrete descendants.

close()

Closes a running system.

class eventsourcing.application.system.SingleThreadedRunner(system: eventsourcing.application.system.System, infrastructure_class=None, *args, **kwargs)

Bases: eventsourcing.application.system.InProcessRunner

Runs a system in the current thread.

__init__(system: eventsourcing.application.system.System, infrastructure_class=None, *args, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

handle_prompt(prompt)

Handles publication of a prompt.

Abstract method which must be implemented on concrete descendants.

run_followers(prompt)

First caller adds a prompt to queue and runs followers until there are no more pending prompts.

Subsequent callers just add a prompt to the queue, avoiding recursion.

class eventsourcing.application.system.MultiThreadedRunner(system: eventsourcing.application.system.System, poll_interval=None, clock_speed=None, **kwargs)

Bases: eventsourcing.application.system.InProcessRunner

Runs a system with a thread for each process.

__init__(system: eventsourcing.application.system.System, poll_interval=None, clock_speed=None, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

start()

Starts running the system.

Abstract method which must be implemented on concrete descendants.

handle_prompt(prompt)

Handles publication of a prompt.

Abstract method which must be implemented on concrete descendants.

close()

Closes a running system.

class eventsourcing.application.system.PromptQueuedApplicationThread(process: eventsourcing.application.process.ProcessApplication, poll_interval=5, inbox=None, outbox=None, clock_event=None)

Bases: threading.Thread

Application thread which uses queues of prompts.

It loops on an “inbox” queue of prompts, and adds its prompts to an “outbox” queue.

__init__(process: eventsourcing.application.process.ProcessApplication, poll_interval=5, inbox=None, outbox=None, clock_event=None)

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is the argument tuple for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

class eventsourcing.application.system.PromptOutbox

Bases: object

Has a collection of downstream prompt inboxes.

__init__()

Initialize self. See help(type(self)) for accurate signature.

put(prompt)

Puts prompt in each downstream inbox (an actual queue).

class eventsourcing.application.system.SteppingRunner(normal_speed=1, scale_factor=1, is_verbose=False, *args, **kwargs)

Bases: eventsourcing.application.system.InProcessRunner

__init__(normal_speed=1, scale_factor=1, is_verbose=False, *args, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

class eventsourcing.application.system.SteppingSingleThreadedRunner(*args, **kwargs)

Bases: eventsourcing.application.system.SteppingRunner

__init__(*args, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

start()

Starts running the system.

Abstract method which must be implemented on concrete descendants.

handle_prompt(prompt)

Ignores prompts.

close()

Closes a running system.

class eventsourcing.application.system.ClockThread(*args, **kwargs)

Bases: threading.Thread

__init__(*args, **kwargs)

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is the argument tuple for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

class eventsourcing.application.system.ProcessRunningClockThread(normal_speed, scale_factor, stop_event: threading.Event, is_verbose=False, seen_prompt_events=None, processes=None, use_direct_query_if_available=False)

Bases: eventsourcing.application.system.ClockThread

__init__(normal_speed, scale_factor, stop_event: threading.Event, is_verbose=False, seen_prompt_events=None, processes=None, use_direct_query_if_available=False)

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is the argument tuple for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

class eventsourcing.application.system.SteppingMultiThreadedRunner(*args, **kwargs)

Bases: eventsourcing.application.system.SteppingRunner

Has a clock thread, and a thread for each application process in the system. The clock thread loops until stopped, waiting for a barrier, after sleeping for remaining tick interval timer. Application threads loop until stopped, waiting for the same barrier. Then, after all threads are waiting at the barrier, the barrier is lifted. The clock thread proceeds by sleeping for the clock tick interval. The application threads proceed by getting new notifications and processing all of them.

There are actually two barriers, so that each application thread waits before getting notifications, and then waits for all processes to complete getting notification before processing the notifications through the application policy. This avoids events created by a process application “bleeding” into the notifications of another process application in the same clock cycle.

Todo: Receive prompts, but set an event for the prompting process, to avoid unnecessary runs.

Allow commands to be scheduled at future clock tick number, and execute when reached.

__init__(*args, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

handle_prompt(prompt)

Handles publication of a prompt.

Abstract method which must be implemented on concrete descendants.

start()

Starts running the system.

Abstract method which must be implemented on concrete descendants.

close()

Closes a running system.

class eventsourcing.application.system.BarrierControlledApplicationThread(process: eventsourcing.application.process.ProcessApplication, fetch_barrier: threading.Barrier, execute_barrier: threading.Barrier, stop_event: threading.Event)

Bases: threading.Thread

__init__(process: eventsourcing.application.process.ProcessApplication, fetch_barrier: threading.Barrier, execute_barrier: threading.Barrier, stop_event: threading.Event)

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is the argument tuple for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

class eventsourcing.application.system.BarrierControllingClockThread(normal_speed, scale_factor, tick_interval, fetch_barrier: threading.Barrier, execute_barrier: threading.Barrier, stop_event: threading.Event, is_verbose=False)

Bases: eventsourcing.application.system.ClockThread

__init__(normal_speed, scale_factor, tick_interval, fetch_barrier: threading.Barrier, execute_barrier: threading.Barrier, stop_event: threading.Event, is_verbose=False)

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is the argument tuple for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

multiprocess

class eventsourcing.application.multiprocess.MultiprocessRunner(system: eventsourcing.application.system.System, pipeline_ids=(0, ), poll_interval=None, setup_tables=False, sleep_for_setup_tables=0, *args, **kwargs)

Bases: eventsourcing.application.system.SystemRunner

__init__(system: eventsourcing.application.system.System, pipeline_ids=(0, ), poll_interval=None, setup_tables=False, sleep_for_setup_tables=0, *args, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

start()

Starts running the system.

Abstract method which must be implemented on concrete descendants.

close()

Closes a running system.

class eventsourcing.application.multiprocess.OperatingSystemProcess(application_process_class, infrastructure_class, upstream_names, pipeline_id=0, poll_interval=5, setup_tables=False, inbox=None, outbox=None, *args, **kwargs)

Bases: multiprocessing.context.Process

__init__(application_process_class, infrastructure_class, upstream_names, pipeline_id=0, poll_interval=5, setup_tables=False, inbox=None, outbox=None, *args, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

run()

Method to be run in sub-process; can be overridden in sub-class

actors

class eventsourcing.application.actors.ActorModelRunner(system: eventsourcing.application.system.System, pipeline_ids, system_actor_name='system', shutdown_on_close=False, **kwargs)

Bases: eventsourcing.application.system.SystemRunner

Uses actor model framework to run a system of process applications.

__init__(system: eventsourcing.application.system.System, pipeline_ids, system_actor_name='system', shutdown_on_close=False, **kwargs)

Initialize self. See help(type(self)) for accurate signature.

start()

Starts all the actors to run a system of process applications.

close()

Stops all the actors running a system of process applications.

class eventsourcing.application.actors.SystemActor

Bases: thespian.actors.Actor

__init__()

Called to initialize the Actor.

Override this initialization method as needed in defined Actors.

N.B. Currently the Actor is not yet fully realized in the ActorSystem when __init__ is invoked. This means that the Actor __init__ cannot invoke any ActorSystem-related operations (no .send(), .handleDeadLetters(), .notifyOnSystemRegistrationChanges(), etc.)

Also note that there is post-__init__ processing of a created Actor object by the ActorSystem that is necessary for it to become a full Actor. The Actor’s __init__() must not perform Actor-related operations, and the __init__() is not sufficient to fully initialize an Actor object. This ensures that the ActorSystem is involved in the creation of a useable Actor (i.e. the ActorSystem is the Factory for an Actor).

receiveMessage(msg, sender)

Main entry point handling a request received by this Actor. Runs without interruption and may access locals to this Actor (only) without concern that these locals will be modified externally.

class eventsourcing.application.actors.PipelineActor

Bases: thespian.actors.Actor

__init__()

Called to initialize the Actor.

Override this initialization method as needed in defined Actors.

N.B. Currently the Actor is not yet fully realized in the ActorSystem when __init__ is invoked. This means that the Actor __init__ cannot invoke any ActorSystem-related operations (no .send(), .handleDeadLetters(), .notifyOnSystemRegistrationChanges(), etc.)

Also note that there is post-__init__ processing of a created Actor object by the ActorSystem that is necessary for it to become a full Actor. The Actor’s __init__() must not perform Actor-related operations, and the __init__() is not sufficient to fully initialize an Actor object. This ensures that the ActorSystem is involved in the creation of a useable Actor (i.e. the ActorSystem is the Factory for an Actor).

receiveMessage(msg, sender)

Main entry point handling a request received by this Actor. Runs without interruption and may access locals to this Actor (only) without concern that these locals will be modified externally.

class eventsourcing.application.actors.ProcessMaster

Bases: thespian.actors.Actor

__init__()

Called to initialize the Actor.

Override this initialization method as needed in defined Actors.

N.B. Currently the Actor is not yet fully realized in the ActorSystem when __init__ is invoked. This means that the Actor __init__ cannot invoke any ActorSystem-related operations (no .send(), .handleDeadLetters(), .notifyOnSystemRegistrationChanges(), etc.)

Also note that there is post-__init__ processing of a created Actor object by the ActorSystem that is necessary for it to become a full Actor. The Actor’s __init__() must not perform Actor-related operations, and the __init__() is not sufficient to fully initialize an Actor object. This ensures that the ActorSystem is involved in the creation of a useable Actor (i.e. the ActorSystem is the Factory for an Actor).

receiveMessage(msg, sender)

Main entry point handling a request received by this Actor. Runs without interruption and may access locals to this Actor (only) without concern that these locals will be modified externally.

class eventsourcing.application.actors.ProcessSlave

Bases: thespian.actors.Actor

__init__()

Called to initialize the Actor.

Override this initialization method as needed in defined Actors.

N.B. Currently the Actor is not yet fully realized in the ActorSystem when __init__ is invoked. This means that the Actor __init__ cannot invoke any ActorSystem-related operations (no .send(), .handleDeadLetters(), .notifyOnSystemRegistrationChanges(), etc.)

Also note that there is post-__init__ processing of a created Actor object by the ActorSystem that is necessary for it to become a full Actor. The Actor’s __init__() must not perform Actor-related operations, and the __init__() is not sufficient to fully initialize an Actor object. This ensures that the ActorSystem is involved in the creation of a useable Actor (i.e. the ActorSystem is the Factory for an Actor).

receiveMessage(msg, sender)

Main entry point handling a request received by this Actor. Runs without interruption and may access locals to this Actor (only) without concern that these locals will be modified externally.