Scatter–gather (fan-out / fan-in) with `resolve_callback_worker`¶

This example shows how to answer one request by fanning it out into many parallel sub-tasks and aggregating their results into a single response — a map-reduce in miniature.

The clever part is where the aggregation happens. Every event a worker sends back arrives at the manager's resolve_callback_worker. By overriding it, the manager becomes the gather point: it collects partial results and, only when the last one arrives, completes the original request.

the manager receives one API request
it scatters N sub-tasks to N workers — and deliberately does not set a response
each worker computes its part and emits it back
resolve_callback_worker gathers the parts; on the last one it fills the original request's response and sends it
the client gets a single aggregated answer

The toy task sums the squares 1² + 2² + … + n², one worker per term. Replace the worker body with whatever you actually fan out (querying N shards, calling N upstreams, processing N file chunks).

Why this works¶

Two framework mechanics make the pattern possible:

A handler that sets no response is not auto-answered. After a @request handler runs, the manager only sends the event back if the handler changed event.response. The scatter handler leaves the response untouched, so the request simply stays pending at the Broker — there is nothing to answer yet.
The original event still carries its correlation identity. Keep a reference to it; when the last part arrives, set its response and put it on the router queue. It is the same object the Broker is waiting on, so it resolves that client's request — even with many requests in flight.

WorkerQuit (sent automatically when each worker finishes) is handled by the framework before resolve_callback_worker, so your override only ever sees the partial-result events. Call super().resolve_callback_worker(...) for anything you don't recognize.

Events¶

# events/squares.py
from http import HTTPMethod

from edri.api.dataclass.api_event import api, ApiType
from edri.dataclass.event import Event, event
from edri.dataclass.response import Response, response


@response
class SumResponse(Response):
    total: int
    parts: int


@api(url="/sum-of-squares/{n}", method=HTTPMethod.GET, resource="sum-of-squares",
     exclude={ApiType.HTML, ApiType.WS})
class SumOfSquares(Event):
    n: int
    response: SumResponse


# Internal carrier: one sub-task in, the same event (with result) back.
@event
class SquarePart(Event):
    job_id: str       # correlates the part to its scatter job
    value: int        # input
    result: int = 0   # filled by the worker

Worker¶

Each worker computes a single term and sends its SquarePart back. A worker must call event_send(...) — modifying the event is not enough.

# workers/squares.py
from edri.abstract.worker import WorkerProcess

from events.squares import SquarePart


class SquareWorker(WorkerProcess[SquarePart]):

    def do(self) -> None:
        self.event.result = self.event.value * self.event.value
        self.event_send(self.event)

Manager (scatter + gather)¶

# managers/squares.py
from uuid import uuid4

from edri.abstract import ManagerBase, request
from edri.dataclass.event import Event
from edri.abstract.worker import Worker

from events.squares import SumOfSquares, SquarePart
from workers.squares import SquareWorker


class FanOutManager(ManagerBase):

    def after_start(self) -> None:
        # job_id -> {"expected": int, "parts": list[int], "origin": SumOfSquares}
        self._jobs: dict = {}

    @request
    def sum_of_squares(self, event: SumOfSquares) -> None:
        # SCATTER: one worker per term, keyed by a fresh job id.
        job_id = uuid4().hex
        values = list(range(1, event.n + 1))
        self._jobs[job_id] = {"expected": len(values), "parts": [], "origin": event}
        for value in values:
            self.workers.start(SquarePart(job_id=job_id, value=value), SquareWorker)
        # Deliberately NO response here -> the request stays pending until gather.

    def resolve_callback_worker(self, event: Event, worker: Worker) -> None:
        # GATHER: collect parts; complete the original request on the last one.
        if isinstance(event, SquarePart):
            job = self._jobs.get(event.job_id)
            if job is None:
                return  # unknown / already-completed job
            job["parts"].append(event.result)
            if len(job["parts"]) == job["expected"]:
                self._jobs.pop(event.job_id)
                origin = job["origin"]
                origin.response.total = sum(job["parts"])
                origin.response.parts = len(job["parts"])
                self.router_queue.put(origin)   # resolves the waiting client
            return
        # Anything else (e.g. results of other patterns) keeps default behavior.
        super().resolve_callback_worker(event, worker)

To cap how many sub-tasks run at once, give the manager a worker limit — FanOutManager(max_workers=8). The framework queues the rest and drains them as workers finish; the gather logic is unchanged.

Entry point¶

# main.py
from edri import EDRI

from managers.squares import FanOutManager

if __name__ == "__main__":
    app = EDRI()
    app.add_component(FanOutManager())
    app.start_api()
    app.run()

Calling it¶

curl -H "Accept: application/json" http://localhost:8080/sum-of-squares/5
# {"data": {"total": 55, "parts": 5}, "status": "OK"}

curl -H "Accept: application/json" http://localhost:8080/sum-of-squares/10
# {"data": {"total": 385, "parts": 10}, "status": "OK"}

Things to watch¶

Response timeout. All parts must arrive within EDRI_API_RESPONSE_TIMEOUT (default 60 s), or the Broker gives up on the client even though the job eventually finishes. Size the fan-out accordingly.
A part that never arrives. If a worker dies without sending its result, the job count never reaches expected and the entry leaks. For production, also count WorkerQuit (override the manager's worker-removal path) or attach a Scheduler-driven timeout that fails the job and answers with an error status.
CPU vs I/O. WorkerProcess gives true parallelism for CPU-bound parts; for I/O-bound fan-out (N network calls) prefer WorkerThread. See Worker.

Scatter–gather (fan-out / fan-in) with resolve_callback_worker¶