AmqpLibReactorTransport
in package
implements
ReactorTransport
The `php-amqplib` implementation of {@see ReactorTransport} (CONTRACT.md §22.1, §8b).
It declares nothing: self::consume() attaches to a queue the server
already declared, and self::publishReply() publishes to the default
exchange, which exists on every broker and needs no declaration. No declare or
bind call appears anywhere in this class, and ReactorRegistryTest asserts that
with a source scan as well as through the shape of the interface.
TLS is mandatory (§8b): self::connect() refuses anything that is not an
amqps:// URL rather than downgrading, because a fallback that works is a
fallback that gets used. Peer and peer-name verification are always on and there
is no switch anywhere in this SDK that turns them off — HMAC does not substitute
for TLS and TLS does not substitute for HMAC.
RECONNECTION IS THE SUPERVISOR'S JOB, exactly as it already is for this SDK's §8
consumer: php-amqplib has no built-in automatic reconnection, so when the
session ends self::wait() returns false, the runtime returns, and the
worker process exits for systemd / Kubernetes / supervisord to restart. A
heartbeat is negotiated so a half-open connection becomes a closed session
rather than a reactor sitting silently attached to a socket nobody is on the
other end of.
Table of Contents
Interfaces
- ReactorTransport
- One live broker session for a reactor: a consumer on the reactor's own queue, and a way to publish a reply back to the queue the delivery named (CONTRACT.md §22.1, §8b).
Constants
- DEFAULT_AMQPS_PORT : mixed = 5671
- The AMQPS port a URL that names none falls back to.
Methods
- __construct() : mixed
- close() : void
- Closes the channel and the connection. Idempotent (§18.1 rule 2).
- connect() : self
- Connects to `$url` over TLS and opens the session channel.
- consume() : void
- Attaches a consumer to `$queue` — the queue the SERVER declared for this reactor. Nothing is declared or bound here (§22.1).
- parseAmqpsUrl() : array{host: string, port: int, user: string, pass: string, vhost: string}
- Splits an `amqps://` URL into the parts `php-amqplib` wants, refusing anything else (§8b).
- publishReply() : void
- Publishes `$body` to `$replyQueue` through the default exchange — the one publication a reactor makes.
- wait() : bool
- Blocks for at most `$timeoutSeconds`, returning false once the session has ended.
Constants
DEFAULT_AMQPS_PORT
The AMQPS port a URL that names none falls back to.
public
mixed
DEFAULT_AMQPS_PORT
= 5671
Methods
__construct()
public
__construct(AbstractConnection $connection, AMQPChannel $channel) : mixed
Parameters
- $connection : AbstractConnection
-
The live broker connection this transport owns and closes.
- $channel : AMQPChannel
-
The session channel. Its QoS is the caller's to set; self::connect() does it.
close()
Closes the channel and the connection. Idempotent (§18.1 rule 2).
public
close() : void
connect()
Connects to `$url` over TLS and opens the session channel.
public
static connect(string $url[, string|null $caFile = null ][, int $heartbeat = 10 ][, int $prefetch = 1 ]) : self
Parameters
- $url : string
-
An
amqps://URL. Anything else is refused (§8b). - $caFile : string|null = null
-
PEM CA bundle to verify the broker against. Omitting it uses the host's trust store. This is the ONLY TLS-related knob: there is no option here or anywhere else in this SDK that weakens or disables verification.
- $heartbeat : int = 10
-
AMQP heartbeat interval in seconds. The heartbeat is how a half-open connection becomes a closed session, which is what lets the supervisor restart the worker instead of leaving it deaf.
- $prefetch : int = 1
-
QoS prefetch. Low by default: a reactor's dispatch window is at most five seconds, so buffering deliveries only guarantees late answers.
Tags
Return values
selfconsume()
Attaches a consumer to `$queue` — the queue the SERVER declared for this reactor. Nothing is declared or bound here (§22.1).
public
consume(string $queue, callable(ReactorDelivery): void $onDelivery) : void
Parameters
- $queue : string
- $onDelivery : callable(ReactorDelivery): void
parseAmqpsUrl()
Splits an `amqps://` URL into the parts `php-amqplib` wants, refusing anything else (§8b).
public
static parseAmqpsUrl(string $url) : array{host: string, port: int, user: string, pass: string, vhost: string}
A plaintext amqp:// URL is refused rather than downgraded. Exposed as a
separate method so the refusal is directly testable without a broker.
Parameters
- $url : string
Tags
Return values
array{host: string, port: int, user: string, pass: string, vhost: string}publishReply()
Publishes `$body` to `$replyQueue` through the default exchange — the one publication a reactor makes.
public
publishReply(string $replyQueue, string $correlationId, string $body) : void
Parameters
- $replyQueue : string
- $correlationId : string
- $body : string
wait()
Blocks for at most `$timeoutSeconds`, returning false once the session has ended.
public
wait(float $timeoutSeconds) : bool
A timeout is an idle tick, not an end: it returns true so the runtime can check whether it has been asked to stop and then wait again.
Parameters
- $timeoutSeconds : float