orxtra v0.13.0 /services.src.orxtra.services._dispatch_worker
Updated
On this page

create_dispatch_worker() wires a DispatchWorker with PgDispatchBackend, ServicesActionExecutor, and AsyncioFlushScheduler, plus a resolver mapping source slugs to their minted principals.

#services.src.orxtra.services._dispatch_worker

#services.src.orxtra.services._dispatch_worker

Factory for constructing a DispatchWorker with concrete implementations.

Bridges the dispatch layer's protocol-based DispatchWorker with the services layer's concrete implementations: PgDispatchBackend, ServicesActionExecutor, AsyncioFlushScheduler.

#_make_source_principal_resolver

python
def _make_source_principal_resolver(backend: PgDispatchBackend, principal_storage: PgPrincipalStorage) -> SourcePrincipalResolver

Build a resolver: source slugs -> source-principal ids.

Each slug is looked up as a source; the source's principal (minted at source birth as KIND_SOURCE/external_ref=source.id) supplies the id. Slugs with no source, or sources without a principal, contribute nothing.

#create_dispatch_worker

python
async def create_dispatch_worker(pool: asyncpg.Pool, *, notification_port: NotificationPort | None=None, cursor_name: str='main', poll_interval: float=5.0, batch_size: int=100) -> DispatchWorker

Construct a DispatchWorker with all concrete service implementations.

Args:

  • pool: asyncpg connection pool.
  • notification_port: optional notification delivery port for NotifyAction.
  • cursor_name: name for the durable cursor (supports multiple workers).
  • poll_interval: fallback poll interval in seconds.
  • batch_size: max events per poll.

Returns:

  • A fully-wired DispatchWorker ready to run().
Search