system

The system layer brings together different process applications within a system definition, and provides various system runners for running a system.

definition

class eventsourcing.system.definition.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.

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

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

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

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

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

Parameters:pipeline_exprs – Pipeline expressions involving process application

classes.

construct_app(process_class: Type[TProcessApplication], infrastructure_class: Optional[Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure]] = None, **kwargs) → TProcessApplication

Constructs process application from given process_class.

__enter__() → eventsourcing.system.definition.AbstractSystemRunner

Supports running a system object directly as a context manager.

The system is run with the SingleThreadedRunner.

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None

Supports usage of a system object as a context manager.

bind(infrastructure_class: Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure]) → TSystem

Constructs a system object that has an infrastructure class from a 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.system.definition.AbstractSystemRunner(system: eventsourcing.system.definition.System, infrastructure_class: Optional[Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure]] = None, setup_tables: bool = False, use_direct_query_if_available: bool = False)

Bases: abc.ABC

__init__(system: eventsourcing.system.definition.System, infrastructure_class: Optional[Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure]] = None, setup_tables: bool = False, use_direct_query_if_available: bool = False)

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

__enter__() → TSystemRunner

Supports usage of a system runner as a context manager.

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None

Supports usage of a system runner as a context manager.

start() → None

Starts running the system.

Abstract method which must be implemented on concrete descendants.

close() → None

Closes a running system.

runner

class eventsourcing.system.runner.InProcessRunner(system: eventsourcing.system.definition.System, infrastructure_class: Optional[Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure]] = None, setup_tables: bool = False, use_direct_query_if_available: bool = False)

Bases: eventsourcing.system.definition.AbstractSystemRunner

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

start() → None

Starts running the system.

Abstract method which must be implemented on concrete descendants.

handle_prompt(prompt: eventsourcing.application.simple.Prompt) → None

Handles publication of a prompt.

Abstract method which must be implemented on concrete descendants.

close() → None

Closes a running system.

class eventsourcing.system.runner.SingleThreadedRunner(system: eventsourcing.system.definition.System, infrastructure_class: Optional[Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure]] = None, **kwargs)

Bases: eventsourcing.system.runner.InProcessRunner

Runs a system in the current thread.

__init__(system: eventsourcing.system.definition.System, infrastructure_class: Optional[Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure]] = None, **kwargs)

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

handle_prompt(prompt: eventsourcing.application.simple.Prompt) → None

Handles publication of a prompt.

Abstract method which must be implemented on concrete descendants.

run_followers(prompt: eventsourcing.application.simple.Prompt) → None

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.system.runner.MultiThreadedRunner(system: eventsourcing.system.definition.System, poll_interval: Optional[int] = None, clock_speed: Union[int, float, None] = None, **kwargs)

Bases: eventsourcing.system.runner.InProcessRunner

Runs a system with a thread for each process.

__init__(system: eventsourcing.system.definition.System, poll_interval: Optional[int] = None, clock_speed: Union[int, float, None] = None, **kwargs)

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

start() → None

Starts running the system.

Abstract method which must be implemented on concrete descendants.

handle_prompt(prompt: eventsourcing.application.simple.Prompt) → None

Handles publication of a prompt.

Abstract method which must be implemented on concrete descendants.

close() → None

Closes a running system.

class eventsourcing.system.runner.PromptOutbox

Bases: typing.Generic

Has a collection of downstream prompt inboxes.

__init__() → None

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

put(prompt: Union[eventsourcing.application.simple.PromptToPull, str]) → None

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

class eventsourcing.system.runner.PromptQueuedApplicationThread(*, process: eventsourcing.application.process.ProcessApplication, poll_interval: int = 5, inbox: queue.Queue, outbox: Optional[eventsourcing.system.runner.PromptOutbox], clock_event: Optional[threading.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: int = 5, inbox: queue.Queue, outbox: Optional[eventsourcing.system.runner.PromptOutbox], clock_event: Optional[threading.Event] = None)

Initialises the thread with a process application object, and inbox and outbox.

Parameters:

• process (ProcessApplication) – Application object.
• poll_interval (int) – Interval to check for upstream events.
• inbox (Queue) – For incoming prompts.
• outbox (PromptOutbox) – For outgoing prompts.
• clock_event – Event that “clocks” this thread (optional).

run() → None

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.system.runner.SteppingRunner(normal_speed: int = 1, scale_factor: int = 1, is_verbose: bool = False, *args, **kwargs)

Bases: eventsourcing.system.runner.InProcessRunner

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

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

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

Bases: eventsourcing.system.runner.SteppingRunner

__init__(*args, **kwargs)

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

start() → None

Starts running the system.

Abstract method which must be implemented on concrete descendants.

handle_prompt(prompt: eventsourcing.application.simple.Prompt) → None

Ignores prompts.

close() → None

Closes a running system.

class eventsourcing.system.runner.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.system.runner.ProcessRunningClockThread(*, normal_speed: int, scale_factor: int, stop_event: threading.Event, is_verbose: bool = False, seen_prompt_events: Dict[str, threading.Event], processes: Dict[str, eventsourcing.application.process.ProcessApplication], use_direct_query_if_available: bool = False)

Bases: eventsourcing.system.runner.ClockThread

__init__(*, normal_speed: int, scale_factor: int, stop_event: threading.Event, is_verbose: bool = False, seen_prompt_events: Dict[str, threading.Event], processes: Dict[str, eventsourcing.application.process.ProcessApplication], use_direct_query_if_available: bool = 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() → None

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.system.runner.SteppingMultiThreadedRunner(*args, **kwargs)

Bases: eventsourcing.system.runner.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: eventsourcing.application.simple.Prompt) → None

Handles publication of a prompt.

Abstract method which must be implemented on concrete descendants.

start() → None

Starts running the system.

Abstract method which must be implemented on concrete descendants.

close() → None

Closes a running system.

class eventsourcing.system.runner.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() → None

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.system.runner.BarrierControllingClockThread(normal_speed: int, scale_factor: int, tick_interval: Union[int, float, None], fetch_barrier: threading.Barrier, execute_barrier: threading.Barrier, stop_event: threading.Event, is_verbose: bool = False)

Bases: eventsourcing.system.runner.ClockThread

__init__(normal_speed: int, scale_factor: int, tick_interval: Union[int, float, None], fetch_barrier: threading.Barrier, execute_barrier: threading.Barrier, stop_event: threading.Event, is_verbose: bool = 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() → None

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.system.multiprocess.MultiprocessRunner(system: eventsourcing.system.definition.System, pipeline_ids: Sequence[int] = (0, ), poll_interval: Optional[int] = None, setup_tables: bool = False, sleep_for_setup_tables: int = 0, **kwargs)

Bases: eventsourcing.system.definition.AbstractSystemRunner

__init__(system: eventsourcing.system.definition.System, pipeline_ids: Sequence[int] = (0, ), poll_interval: Optional[int] = None, setup_tables: bool = False, sleep_for_setup_tables: int = 0, **kwargs)

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

start() → None

Starts running the system.

Abstract method which must be implemented on concrete descendants.

close() → None

Closes a running system.

class eventsourcing.system.multiprocess.OperatingSystemProcess(application_process_class: Type[eventsourcing.application.process.ProcessApplication], infrastructure_class: Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure], upstream_names: List[str], inbox: queue.Queue, outbox: Optional[eventsourcing.system.runner.PromptOutbox[typing.Tuple[int, str]][Tuple[int, str]]] = None, pipeline_id: int = 0, poll_interval: int = 5, setup_tables: bool = False, *args, **kwargs)

Bases: multiprocessing.context.Process

__init__(application_process_class: Type[eventsourcing.application.process.ProcessApplication], infrastructure_class: Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure], upstream_names: List[str], inbox: queue.Queue, outbox: Optional[eventsourcing.system.runner.PromptOutbox[typing.Tuple[int, str]][Tuple[int, str]]] = None, pipeline_id: int = 0, poll_interval: int = 5, setup_tables: bool = False, *args, **kwargs)

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

run() → None

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

actors

class eventsourcing.system.thespian.ThespianRunner(system: eventsourcing.system.definition.System, pipeline_ids=(0, ), system_actor_name='system', shutdown_on_close=False, **kwargs)

Bases: eventsourcing.system.definition.AbstractSystemRunner

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

__init__(system: eventsourcing.system.definition.System, pipeline_ids=(0, ), 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.system.thespian.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.system.thespian.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.system.thespian.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.system.thespian.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.