GenServer-based async event dispatcher for the agent system.

The AsyncDispatcher manages event routing between agents in an asynchronous, non-blocking manner. It maintains an event queue (FIFO) and processes events by routing them through a Mojentic.Router to the appropriate agents.

Features

• Event Queue - FIFO queue using :queue module
• Async Processing - Non-blocking event handling via Task and GenServer
• Mixed Agent Support - Handles both sync and async agents
• Graceful Shutdown - Stop via TerminateEvent or explicit stop/1
• Queue Monitoring - Wait for empty queue with wait_for_empty_queue/2

Architecture

┌─────────────┐
│  Dispatcher │
│   (GenServer) │
└──────┬──────┘
       │ Event Queue
       │ [:queue]
       │
       ├─→ Router ─→ Agent1 ──→ [New Events]
       │            ↓
       └─→ Router ─→ Agent2 ──→ [New Events]

State Structure

%{
  router: %Router{},
  event_queue: :queue.queue(),
  processing: boolean(),
  batch_size: integer()
}

Usage

# Create router
router = Router.new()
|> Router.add_route(QuestionEvent, fact_checker)
|> Router.add_route(QuestionEvent, answer_generator)

# Start dispatcher
{:ok, pid} = AsyncDispatcher.start_link(router: router)

# Dispatch events
event = %QuestionEvent{source: MyApp, question: "What is Elixir?"}
AsyncDispatcher.dispatch(pid, event)

# Wait for queue to empty
:ok = AsyncDispatcher.wait_for_empty_queue(pid, timeout: 10_000)

# Stop dispatcher
AsyncDispatcher.stop(pid)

Examples

# Full workflow
router = Router.new()
|> Router.add_route(QuestionEvent, fact_checker_pid)
|> Router.add_route(FactCheckEvent, aggregator_pid)

{:ok, dispatcher} = AsyncDispatcher.start_link(
  router: router,
  batch_size: 10
)

question = %QuestionEvent{
  source: MyApp,
  question: "What is the capital of France?"
}

AsyncDispatcher.dispatch(dispatcher, question)
AsyncDispatcher.wait_for_empty_queue(dispatcher)
AsyncDispatcher.stop(dispatcher)

Concurrency Model

Agents matching an event are dispatched concurrently via Tasks. This is intentional and is safe in Elixir because processes do not share mutable memory. The other mojentic ports (ts/py/ru) serialize agent dispatch precisely because they DO share mutable state (SharedWorkingMemory etc.) where concurrent agents would race.