Transports

For a framework-less PHP producer, the core ships optional reference Transport implementations covering all seven brokers — Redis, RabbitMQ, Amazon SQS, Apache Kafka, Apache Pulsar and Apache ActiveMQ/Artemis. (Azure Service Bus is the one broker without a PHP transport — ADR-0021.) They keep the core dependency-free — install only the broker client you use.

The Transport seam

BabelQueue\Contracts\Transport is a single method:

interface Transport
{
    public function publish(string $payload, ?string $queue = null): ?string;
}

$payload is the EnvelopeCodec::encode(...) output. Any broker client can implement it — phpredis (ext-redis) users, for example, can do it in one line (rpush).

RedisTransport

composer require predis/predis

use BabelQueue\Codec\EnvelopeCodec;
use BabelQueue\Transport\RedisTransport;

$transport = new RedisTransport(new Predis\Client('redis://localhost:6379'));

$envelope = EnvelopeCodec::fromJob($job, 'orders');
$transport->publish(EnvelopeCodec::encode($envelope), 'orders');
// a Go / Python / Node consumer reads the identical envelope off "orders"

RedisTransport produces with RPUSH, matching the reliable-queue pattern the other SDKs use.

AmqpTransport

composer require php-amqplib/php-amqplib

use BabelQueue\Transport\AmqpTransport;

$transport = new AmqpTransport(/* php-amqplib connection */);
$transport->publish(EnvelopeCodec::encode($envelope), 'orders');

AmqpTransport declares a durable queue, publishes persistent messages, and sets the contract AMQP properties (type = URN, correlation_id = trace_id, message_id = meta.id, plus x-schema-version / x-source-lang / x-attempts), so consumers can route on properties without parsing the body.

SqsTransport

composer require aws/aws-sdk-php

A produce-side Amazon SQS transport: it sends the canonical envelope as the MessageBody with the §3 MessageAttributes. It is decoupled from the AWS SDK behind a one-method BabelQueue\Transport\SqsClient seam, so wrap a real Aws\Sqs\SqsClient in a one-line adapter:

use BabelQueue\Transport\SqsClient;
use BabelQueue\Transport\SqsTransport;

$adapter = new class (new Aws\Sqs\SqsClient([/* region, credentials */])) implements SqsClient {
    public function __construct(private Aws\Sqs\SqsClient $c) {}
    public function sendMessage(array $args): mixed { return $this->c->sendMessage($args); }
};

$transport = new SqsTransport($adapter, 'https://sqs.eu-central-1.amazonaws.com/123456789012/orders');
$transport->publish(EnvelopeCodec::encode($envelope));
// a Go / Python / Node / Java / .NET consumer reads the identical envelope + attributes

The projected attributes (bq-job = URN, bq-trace-id, bq-message-id, plus bq-schema-version / bq-source-lang / bq-created-at) let a consumer route without decoding the body; FIFO sets MessageGroupId / MessageDeduplicationId.

KafkaTransport

Apache Kafka (§6) is PHP’s one opt-in transport: its only viable client is the ext-rdkafka C extension, so it deliberately relaxes the zero-extension rule (ADR-0019). The transport stays decoupled behind the one-method BabelQueue\Transport\KafkaProducer seam, so you wrap your real RdKafka\Producer in a one-line adapter:

pecl install rdkafka   # the ext-rdkafka extension

use BabelQueue\Transport\KafkaProducer;
use BabelQueue\Transport\KafkaTransport;

$producer = new class (/* RdKafka\Producer */) implements KafkaProducer {
    public function produce(string $topic, string $payload, array $headers, ?int $timestampMs = null): void {
        // produce to $topic with the bq-* headers + record timestamp
    }
};

$transport = new KafkaTransport($producer, 'orders');
$transport->publish(EnvelopeCodec::encode($envelope), 'orders');
// a Java / Go / Node / Python / .NET consumer reads the identical envelope off the topic

The record value is the canonical envelope; the contract fields ride bq- record headers (bq-job = URN, bq-trace-id, bq-message-id, bq-schema-version, bq-source-lang, bq-attempts — hyphens, per §6), and the record timestamp mirrors meta.created_at.

PulsarTransport

Apache Pulsar (§5) has no mature native PHP client, so PHP produces over Pulsar’s native WebSocket API with a pure-PHP WebSocket client — GR-7 stays intact (no C extension, ADR-0020). It is decoupled behind the one-method BabelQueue\Transport\PulsarWebSocketClient seam:

composer require textalk/websocket

use BabelQueue\Transport\PulsarTransport;
use BabelQueue\Transport\PulsarWebSocketClient;

$client = new class (/* textalk/websocket client */) implements PulsarWebSocketClient {
    public function publish(string $topic, string $payload, array $properties): void {
        // base64 the payload into a WS producer frame with the bq-* properties; check the ack
    }
};

$transport = new PulsarTransport($client, 'orders'); // tenant/namespace default to public/default
$transport->publish(EnvelopeCodec::encode($envelope), 'orders');
// the BabelQueue "queue" maps to persistent://public/default/orders

The message value is the canonical envelope; bq- native properties (string→string, hyphens) carry the contract fields. The body’s meta.created_at stays authoritative — the adapter must not set the frame’s eventTime.

StompTransport (Artemis)

Apache ActiveMQ Artemis (§7) speaks AMQP 1.0, which has no strong PHP client, so PHP reaches it over STOMP (the mature pure-PHP stomp-php client; Artemis ships a STOMP acceptor) — GR-7 intact (ADR-0018). Artemis bridges STOMP ↔ AMQP 1.0 ↔ JMS on the same address, so a STOMP-produced message is consumed natively by the Java (JMS) and .NET / Node / Python / Go (AMQP 1.0) SDKs:

composer require stomp-php/stomp-php

use BabelQueue\Transport\StompClient;
use BabelQueue\Transport\StompTransport;

$client = new class (/* Stomp\Client */) implements StompClient {
    public function send(string $destination, string $body, array $headers): void {
        // SEND the envelope JSON to the anycast address with the bq_ headers
    }
};

$transport = new StompTransport($client, 'orders');
$transport->publish(EnvelopeCodec::encode($envelope), 'orders');

Routing is body-authoritative (a STOMP header can’t set the x-opt-jms-type annotation, so consumers fall back to the body’s job URN, which is always present). The §7 fields ride correlation-id (= trace_id) and the bq_ application properties — note the underscores (JMS forbids hyphens in property names — ADR-0017).

Consuming with the core. These transports are producers. PHP consumes Kafka and Pulsar with the framework-less KafkaConsumer / PulsarConsumer, and Artemis with the Laravel babelqueue-artemis drop-in driver.

On Laravel or Symfony you don’t need these — the Laravel driver and the Symfony serializer move the bytes through the framework’s own transport.