bobocep.cep.engine.forwarder.forwarder.BoboForwarder

class bobocep.cep.engine.forwarder.forwarder.BoboForwarder(phenomena: List[BoboPhenomenon], handler: BoboActionHandler, gen_event_id: BoboGenEventID, gen_timestamp: BoboGenTimestamp, local_only: bool = True, max_size: int = 0)[source]

Bases: BoboEngineTask, BoboForwarderPublisher, BoboProducerSubscriber

A forwarder task.

__init__(phenomena: List[BoboPhenomenon], handler: BoboActionHandler, gen_event_id: BoboGenEventID, gen_timestamp: BoboGenTimestamp, local_only: bool = True, max_size: int = 0)[source]
Parameters:

  • phenomena – List of phenomena.

  • handler – Action handler.

  • gen_event_id – Event ID generator.

  • gen_timestamp – Timestamp generator.

  • local_only – If True, forwarder only executes actions for locally-completed complex events i.e. not complex events that were completed on a distributed instance. Note: if False, the action may be executed more than once: once by the instance that generated the remote event, and another by this instance.

  • max_size – Maximum queue size. Default: 0 (unbounded).

close() → None[source]

Closes the Forwarder.

is_closed() → bool[source]
Returns:

True if Forwarder is closed; False otherwise.

on_producer_update(event: BoboEventComplex, local: bool) → None[source]
Parameters:

  • event – Complex event generated by Producer.

  • local – True if the complex event was generated using a locally-completed run; False otherwise.

size() → int[source]
Returns:

Queue size.

subscribe(subscriber: BoboForwarderSubscriber)[source]
Parameters:

subscriber – Subscriber to Forwarder data.

update() → bool[source]
Returns:

True if an internal update occurred; False otherwise.