Ecore¶

class ECore(stream_factory, *, process_pipeline=None, process_pipeline_type=<class 'eventscore.core.pipelines.ProcessPipeline'>, process_pipeline_init_kwargs=None, spawn_worker=None, spawn_worker_type=<class 'eventscore.core.workers.SpawnMPWorker'>, spawn_worker_init_kwargs=None, producer=None, producer_type=<class 'eventscore.core.producers.Producer'>, producer_init_kwargs=None, logger=<RootLogger root (DEBUG)>, skipping_predicate=<function _skipping_predicate>)[source]¶

Bases: IECore

Parameters:

stream_factory (IStreamFactory)
process_pipeline (IProcessPipeline | None)
process_pipeline_type (type[IProcessPipeline])
process_pipeline_init_kwargs (dict[str, Any] | None)
spawn_worker (ISpawnWorker | None)
spawn_worker_type (type[ISpawnWorker])
spawn_worker_init_kwargs (dict[str, Any] | None)
producer (IProducer | None)
producer_type (type[IProducer])
producer_init_kwargs (dict[str, Any] | None)
logger (Logger)
skipping_predicate (Callable[[], bool] | None)

property process_pipeline: IProcessPipeline¶

Pipeline processor getter

Returns:: Pipeline processor
Return type:: IProcessPipeline

property spawn_worker: ISpawnWorker¶

Worker spawner getter

Returns:: Worker spawner
Return type:: ISpawnWorker

property producer: IProducer¶

Producer getter

Returns:: Producer
Return type:: IProducer

property stream_factory: IStreamFactory¶

Stream factory getter

Returns:: Stream factory
Return type:: IStreamFactory

property stream: IStream¶

Stream getter

Returns:: Stream instance
Return type:: IStream

consumer(func=None, *, event, group, clones=1)[source]¶

Decorator for consumer functions

Parameters:

func (ConsumerFunc | None) – function to decorate
event (EventType) – Event type
group (ConsumerGroup) – Consumer group
clones (NumberOfClones) – No of clones

Returns:

Decorated function

Return type:

ConsumerFunc

register_consumer(func, event, group, *, clones=1, func_path=None)[source]¶

Consumer function registrator

Parameters:

func (ConsumerFunc) – Function to register as a consumer
event (EventType) – Event type
group (ConsumerGroup) – Consumer group
clones (NumberOfClones) – No of clones
func_path (str | None) – Path (absolute preferred) to module where function is defined. Is used for consumer functions equality check to avoid duplicate registering. Defaults to ` (inspect.getsourcefile(func) or "") + ":" + func.__name__ `

Returns:

None

Return type:

None

discover_consumers(*, root='')[source]¶

Discover consumers within given package root.

Parameters:: root (str | None) – root package to search in. Current directory is used by default.
Returns:: None
Return type:: None

produce(event, *, block=True, timeout=5)[source]¶

Produce an event

Parameters:

event (Event) – Event to produce
block (bool) – Should I/O be blocked if some delay occurs. Defaults to True.
timeout (int) – Number of seconds to wait in case of latency. Defaults to 5.

Returns:

None

Return type:

None

spawn_workers()[source]¶

Spawn workers for registered consumers. Further consumer registering and spawning won’t matter

Returns:: None
Return type:: None