README

Subscribe to OPC UA PubSub streams directly from PHP — UDP unicast or multicast, UADP and JSON payloads, group-key security — without any C/C++ extensions, HTTP gateways, or middleware in between.

This package extends php-opcua/opcua-client with the OPC UA Part 14 — PubSub Subscriber role. The core client stays unchanged; installing this extension only adds the PhpOpcua\Client\ExtTransportPubSub\* namespace. Same zero-dependency philosophy, same PSR-3 / PSR-14 integration, same cross-platform support.

What you can do with it:

• Listen to PubSub NetworkMessages over UDP unicast or multicast
• Decode UADP binary and JSON payloads into typed DataSetMessages with named fields
• Verify and decrypt signed and encrypted PubSub streams using pre-shared group keys or a Security Key Service
• Configure readers by (publisherId, writerGroupId, dataSetWriterId) and let the kernel demux incoming traffic
• React in real time via PSR-14 events or plain callbacks on every decoded field

All this with ext-sockets as the only extension beyond the core's ext-openssl, and PHP 8.2 through 8.5 supported on Linux, macOS, and Windows.

Note: PubSub is a fire-and-forget broadcast paradigm, not a session-oriented protocol. A long-running worker process (ReactPHP, Symfony Messenger, Laravel queue worker, Artisan command, systemd unit) is the natural place to run a Subscriber. Short-lived PHP requests cannot sustain a meaningful subscription window.

Quick Start

composer require php-opcua/opcua-client-ext-transport-pubsub

use PhpOpcua\Client\ExtTransportPubSub\SubscriberBuilder;
use PhpOpcua\Client\ExtTransportPubSub\Types\DataSetMessage;
use PhpOpcua\Client\ExtTransportPubSub\Types\DataSetMetaData;
use PhpOpcua\Client\ExtTransportPubSub\Types\DataSetReaderConfig;

$metadata = DataSetMetaData::fromJsonFile('/etc/opcua/line1.json');

$subscriber = SubscriberBuilder::create()
    ->onDataSetMessage(function (DataSetMessage $msg) {
        foreach ($msg->fields as $field) {
            echo "{$field->name} = {$field->getScalar()}\n";
        }
    })
    ->listenUdp(
        endpoint: 'opc.udp://239.0.0.1:4840',
        readers: [
            new DataSetReaderConfig(
                publisherId: 100,
                writerGroupId: 1,
                dataSetWriterId: 1,
                dataSetMetaData: $metadata,
            ),
        ],
    );

$subscriber->run();

That's it. Build, configure one reader, block on run(). Press Ctrl-C (or call stop() from a signal handler) to exit cleanly.

Tip: Prefer poll(timeoutMs: 500) over run() when you have your own event loop (ReactPHP, Amp, custom). It returns the batch of decoded DataSetMessage objects from this tick without blocking past the timeout.

See It in Action

Listen on unicast UDP

$subscriber = SubscriberBuilder::create()
    ->listenUdp(
        endpoint: 'opc.udp://192.168.1.50:4840',
        readers: [$reader],
    );

Tune the UDP socket

use PhpOpcua\Client\ExtTransportPubSub\Transport\UdpOptions;

$subscriber = SubscriberBuilder::create()
    ->listenUdp(
        endpoint: 'opc.udp://239.0.0.1:4840',
        transport: new UdpOptions(
            interface: '0.0.0.0',
            receiveBufferSize: 131072,
            ttl: 8,
            reuseAddress: true,
        ),
        readers: [$reader],
    );

Decode JSON payloads

$subscriber = SubscriberBuilder::create()
    ->useJson()
    ->listenUdp(
        endpoint: 'opc.udp://239.0.0.1:4840',
        readers: [$reader],
    );

Load metadata from JSON, XML, or a live server

use PhpOpcua\Client\ExtTransportPubSub\Types\DataSetMetaData;

// From a JSON config file
$metadata = DataSetMetaData::fromJsonFile('/etc/opcua/line1.json');

// From an OPC UA DataSetMetaDataType XML file
$metadata = DataSetMetaData::fromXmlFile('/etc/opcua/line1.xml');

// From a running server via the classic opcua-client
$metadata = DataSetMetaData::fetchFromServer($client, 'ns=2;s=PDS/Line1/Metadata');

Secure the stream with pre-shared keys

use PhpOpcua\Client\ExtTransportPubSub\Security\PubSubSecurityMode;
use PhpOpcua\Client\ExtTransportPubSub\Security\PubSubSecurityOptions;
use PhpOpcua\Client\ExtTransportPubSub\Security\StaticGroupKeyProvider;

$subscriber = SubscriberBuilder::create()
    ->listenUdp(
        endpoint: 'opc.udp://239.0.0.1:4840',
        security: new PubSubSecurityOptions(
            mode: PubSubSecurityMode::SignAndEncrypt,
            keyProvider: new StaticGroupKeyProvider(
                signingKey: hex2bin($_ENV['PUBSUB_SIGN']),
                encryptingKey: hex2bin($_ENV['PUBSUB_ENC']),
                keyNonce: hex2bin($_ENV['PUBSUB_NONCE']),
            ),
        ),
        readers: [$reader],
    );

Rotate keys from a Security Key Service

use PhpOpcua\Client\ClientBuilder;
use PhpOpcua\Client\ExtTransportPubSub\Security\SksGroupKeyProvider;

$client = ClientBuilder::create()->connect('opc.tcp://sks.example.com:4840');

$provider = new SksGroupKeyProvider($client, securityGroupId: 'line1-ops');
$provider->refresh();

$subscriber = SubscriberBuilder::create()
    ->listenUdp(
        endpoint: 'opc.udp://239.0.0.1:4840',
        security: new PubSubSecurityOptions(PubSubSecurityMode::SignAndEncrypt, $provider),
        readers: [$reader],
    );

Poll instead of run

while (! $shouldStop) {
    foreach ($subscriber->poll(timeoutMs: 250) as $msg) {
        handle($msg);
    }
    doOtherWork();
}

Graceful shutdown from a signal handler

pcntl_async_signals(true);
pcntl_signal(SIGTERM, fn () => $subscriber->stop());
pcntl_signal(SIGINT, fn () => $subscriber->stop());

$subscriber->run();

Receive PSR-14 events for every step

use PhpOpcua\Client\ExtTransportPubSub\Event\DataSetFieldReceived;
use PhpOpcua\Client\ExtTransportPubSub\Event\MessageDecodeError;
use PhpOpcua\Client\ExtTransportPubSub\Event\SecurityValidationFailed;

$subscriber = SubscriberBuilder::create()
    ->setEventDispatcher($yourDispatcher)
    ->listenUdp(/* ... */);

Listeners fire for NetworkMessageReceived, DataSetMessageReceived, DataSetFieldReceived, MessageDecodeError, SecurityValidationFailed, TransportOpened, TransportClosed, and TransportError. Zero overhead when no dispatcher is configured.

Add structured logging

use Monolog\Logger;
use Monolog\Handler\StreamHandler;

$logger = new Logger('pubsub');
$logger->pushHandler(new StreamHandler('php://stderr', Logger::DEBUG));

$subscriber = SubscriberBuilder::create()
    ->setLogger($logger)
    ->listenUdp(/* ... */);

Any PSR-3 logger works. Without one, logging is silently disabled.

Plug in your own transport

use PhpOpcua\Client\ExtTransportPubSub\Transport\PubSubTransportInterface;

class MyKafkaTransport implements PubSubTransportInterface { /* ... */ }

$subscriber = SubscriberBuilder::create()
    ->listenOn(
        transports: [new MyKafkaTransport(/* ... */)],
        readers: [$reader],
    );

The PubSubTransportInterface contract is the extension point that keeps the subscriber agnostic to the underlying transport. External packages like php-opcua/opcua-client-ext-mqtt plug in by implementing this interface.

Why This Package?

• Zero runtime dependencies beyond the core — ext-sockets plus the core's ext-openssl. Optional PSR-3 logging and PSR-14 events via any compatible implementation.
• PHP 8.2+ — runs on any modern PHP.
• Native UADP binary — speaks the OPC UA binary PubSub encoding directly over UDP. No translation layer.
• UDP unicast and multicast — IGMP group join with MCAST_JOIN_GROUP plus legacy IP_ADD_MEMBERSHIP fallback. TTL, buffer size, and SO_REUSEADDR exposed as first-class options.
• JSON codec included — the reversible form from Part 14 §7.2, round-trippable without external metadata.
• Group-key security — HMAC-SHA256 signing and AES-256-CBC encryption handled in pure PHP via ext-openssl. Pre-shared keys or live rotation from a Security Key Service.
• Metadata any way you like — PHP arrays, JSON files, DataSetMetaDataType XML exports, or a live read from the publishing server.
• Typed everywhere — every DTO uses public readonly properties. No arrays, no magic.
• Cross-platform — tested on Linux, macOS, and Windows across PHP 8.2–8.5. No FFI, no COM, no platform-specific APIs beyond ext-sockets.
• Does not modify the core — ClientBuilder, Client, ClientKernel, OpcUaClientInterface are untouched.

Features

Documentation

Full docs live under docs/ (published at https://www.php-opcua.com/documentation/opcua-client-ext-transport-pubsub).

Testing

Run ./vendor/bin/pest after composer install. Unit tests cover codec round-trips, UDP loopback send/receive, kernel demux, security unwrap, metadata loaders, and SKS key fetching. CI runs on PHP 8.2, 8.3, 8.4, and 8.5 across Linux, macOS, and Windows via GitHub Actions.

./vendor/bin/pest                                          # everything
./vendor/bin/pest tests/Unit/                              # unit only
./vendor/bin/pest tests/Integration/ --group=integration   # integration only

Ecosystem

Community

Have questions, ideas, or want to share what you've built? Join the GitHub Discussions.

Connected a PubSub-capable device? We're collecting a community-driven list of tested publishers. Share your experience — even a one-liner like "open62541 1.4, UDP multicast, works fine" helps other users know what to expect.

AI-Ready

This package ships with machine-readable documentation designed for AI coding assistants (Claude, Cursor, Copilot, ChatGPT, and others). Feed these files to your AI so it knows how to use the library correctly:

How to use: copy the files you need into your project's AI configuration directory. The files are located in vendor/php-opcua/opcua-client-ext-transport-pubsub/ after composer install.

• Claude Code: reference per-session with --add-file vendor/php-opcua/opcua-client-ext-transport-pubsub/llms-skills.md
• Cursor: copy into your project's rules directory — cp vendor/php-opcua/opcua-client-ext-transport-pubsub/llms-skills.md .cursor/rules/opcua-pubsub.md
• GitHub Copilot: copy or append into .github/copilot-instructions.md
• Other tools: paste the content into your system prompt or project knowledge base

Roadmap

See ROADMAP.md for what's coming next.

Contributing

Contributions welcome — see CONTRIBUTING.md.

Changelog

See CHANGELOG.md.

License

MIT