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)[source]¶ 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)[source]¶ Initialises a “process network” system object.
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).
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[source]¶ Constructs process application from given
process_class
.
-
__enter__
() → eventsourcing.system.definition.AbstractSystemRunner[source]¶ 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[source]¶ Supports usage of a system object as a context manager.
-
bind
(infrastructure_class: Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure]) → TSystem[source]¶ 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)[source]¶ 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)[source]¶ Initialize self. See help(type(self)) for accurate signature.
-
__exit__
(exc_type: Any, exc_val: Any, exc_tb: Any) → None[source]¶ Supports usage of a system runner as a context manager.
-
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)[source]¶ 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[source]¶ Starts running the system.
Abstract method which must be implemented on concrete descendants.
-
-
class
eventsourcing.system.runner.
SingleThreadedRunner
(system: eventsourcing.system.definition.System, infrastructure_class: Optional[Type[eventsourcing.application.simple.ApplicationWithConcreteInfrastructure]] = None, **kwargs)[source]¶ 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)[source]¶ Initialize self. See help(type(self)) for accurate signature.
-
-
class
eventsourcing.system.runner.
MultiThreadedRunner
(system: eventsourcing.system.definition.System, poll_interval: Optional[int] = None, clock_speed: Union[int, float, None] = None, **kwargs)[source]¶ 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)[source]¶ Initialize self. See help(type(self)) for accurate signature.
-
start
() → None[source]¶ Starts running the system.
Abstract method which must be implemented on concrete descendants.
-
-
class
eventsourcing.system.runner.
PromptOutbox
[source]¶ Bases:
typing.Generic
Has a collection of downstream prompt inboxes.
-
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)[source]¶ 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)[source]¶ 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[source]¶ 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)[source]¶
-
class
eventsourcing.system.runner.
SteppingSingleThreadedRunner
(*args, **kwargs)[source]¶ Bases:
eventsourcing.system.runner.SteppingRunner
-
start
() → None[source]¶ Starts running the system.
Abstract method which must be implemented on concrete descendants.
-
-
class
eventsourcing.system.runner.
ClockThread
(*args, **kwargs)[source]¶ Bases:
threading.Thread
-
__init__
(*args, **kwargs)[source]¶ 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)[source]¶ 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)[source]¶ 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[source]¶ 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)[source]¶ 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.
-
handle_prompts
(prompts: Iterable[eventsourcing.application.process.Prompt]) → None[source]¶ Handles publication of a prompt.
Abstract method which must be implemented on concrete descendants.
-
-
class
eventsourcing.system.runner.
BarrierControlledApplicationThread
(process: eventsourcing.application.process.ProcessApplication, fetch_barrier: threading.Barrier, execute_barrier: threading.Barrier, stop_event: threading.Event)[source]¶ Bases:
threading.Thread
-
__init__
(process: eventsourcing.application.process.ProcessApplication, fetch_barrier: threading.Barrier, execute_barrier: threading.Barrier, stop_event: threading.Event)[source]¶ 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[source]¶ 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)[source]¶ 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)[source]¶ 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[source]¶ 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.
-
actors¶
-
class
eventsourcing.system.actors.
ActorModelRunner
(system: eventsourcing.system.definition.System, pipeline_ids=(0, ), system_actor_name='system', shutdown_on_close=False, **kwargs)[source]¶ Bases:
eventsourcing.system.definition.AbstractSystemRunner
Uses actor model framework to run a system of process applications.
-
class
eventsourcing.system.actors.
SystemActor
[source]¶ Bases:
thespian.actors.Actor
-
__init__
()[source]¶ 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).
-
-
class
eventsourcing.system.actors.
PipelineActor
[source]¶ Bases:
thespian.actors.Actor
-
__init__
()[source]¶ 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).
-
-
class
eventsourcing.system.actors.
ProcessMaster
[source]¶ Bases:
thespian.actors.Actor
-
__init__
()[source]¶ 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).
-
-
class
eventsourcing.system.actors.
ProcessSlave
[source]¶ Bases:
thespian.actors.Actor
-
__init__
()[source]¶ 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).
-
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)[source]¶ 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)[source]¶ Initialize self. See help(type(self)) for accurate signature.
-
-
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)[source]¶ 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)[source]¶ Initialize self. See help(type(self)) for accurate signature.
-