AXIAM PHP SDK

AmqpLibReactorTransport
in package
implements ReactorTransport

FinalYes

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.

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
throws
InvalidArgumentException

when $url is not amqps://.

Return values
self

consume()

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
throws
InvalidArgumentException

when the URL is not amqps://, or is unparseable.

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
Return values
bool
On this page

Search results