Download the PHP package flyokai/amp-channel-dispatcher without Composer

flyokai/amp-channel-dispatcher

User docs → AGENTS.md

Async bidirectional message dispatching over AMPHP channels — request/response correlation, middleware pipeline, remote iterators, and configurable error handling.

This is the messaging plumbing that lets two PHP processes (or threads) talk to each other over an AMPHP Channel. You get request/response semantics with automatic id correlation, a middleware stack for cross-cutting concerns, and remote iterators for streaming large result sets without buffering.

Features

• Dual-loop dispatcher — read & write loops share one channel
• Request types — Request (expects Response) and MeekRequest (fire-and-forget)
• Middleware pipeline — chain-of-responsibility via stackMiddleware()
• Remote iterators — stream results across the channel without full buffering
• Pluggable error handling — ErrorResponse (recoverable) vs FatalErrorResponse (terminates dispatcher)
• Context injection — handlers get a Context with reference back to the dispatcher

Installation

Quick start

Concepts

Messages

Every message implements Message:

• id(): int — auto-generated unique ID
• getAttribute(string $name) / setAttribute(string $name, mixed $value)
• cloneWith(...$args): static

Request and Response both extend Message. A Response carries requestId(): ?int for correlation.

Built-in responses

Dispatcher lifecycle

The dispatcher attaches a Context attribute to every incoming request with:

• dispatcher() — ?WeakReference to the dispatcher
• sendRequest() — for nested calls
• iterator-storage handles (addLocalIterator(), …)

Middleware

Remote iterators

When a handler returns a ConcurrentIterator, the consumer side gets a RemoteIterator proxy. Each continue() is a blocking RPC round-trip:

API

Gotchas

• WeakReference inside Context — if the dispatcher is GC'd, context->dispatcher() returns null. Handlers must check.
• RemoteIterator round-trips — every continue() blocks until the remote responds. There is no batching; very large iterators are slow.
• FiberLocal in RemoteIterator — state is per-fiber. Calling continue() from a different fiber raises an error.
• MeekRequest — sendRequest() returns null. There is no delivery confirmation.
• stop() cancels pending futures — every unresolved request errors with DispatcherException('Dispatcher terminated').
• FatalErrorResponse halts the dispatcher — a response with a null requestId terminates the read loop.
• Messages are serialized — AMPHP channels use serialize()/unserialize(). Custom objects must support it.
• WeakClosure callbacks — all dispatcher callbacks are weakly referenced to avoid circular references and let GC do its job.

See also

• flyokai/data-service — socket-based service built on this dispatcher
• flyokai/data-service-message — concrete request/response DTOs

License

MIT