only-core

IoT Data Collector — OPC UA scaffold (only-core)

A small scaffold for building an IoT data collector on top of php-opcua/opcua-client, without a session manager, framework, or external storage. Two entry points share the same configuration and the same PSR-14 event bus:

• bin/collector-daemon.php — long-running worker, OPC UA subscriptions, push-style data flow, automatic reconnect.
• bin/collector-cron.php — one-shot readMulti batch, exits with a status code suitable for cron / systemd timers.

The collectors only connect, monitor / read, and emit events. Everything else — persistence, forwarding, alerting — lives in listeners.php, the single user-editable scaffold file.

---

Layout

.
├── bin/
│   ├── collector-daemon.php         # entry point — long-running worker
│   └── collector-cron.php           # entry point — one-shot batch
├── src/
│   ├── Bootstrap.php                # client / dispatcher / tags factory
│   ├── Tag.php                      # tag DTO (name, nodeId, samplingInterval, clientHandle)
│   ├── DaemonCollector.php          # subscription loop, reconnect supervisor
│   ├── CronCollector.php            # readMulti, dispatch, exit
│   ├── Event/TagSampled.php         # unified application event
│   ├── EventDispatcher/
│   │   └── SimpleDispatcher.php     # ~50-line PSR-14 dispatcher + listener provider
│   └── Logger/StderrLogger.php      # PSR-3 logger writing to stderr
├── config.php                       # endpoint, security, subscription, reconnect, tags
├── listeners.php                    # ← user-editable scaffold (PSR-14 listeners)
├── composer.json
├── LICENSE
└── README.md

Requirements

• PHP 8.2+ with ext-openssl (no other extensions required)
• pcntl is used for graceful SIGINT/SIGTERM handling (Linux/macOS daemon use case); on Windows the daemon still runs but Ctrl+C kills it abruptly
• An OPC UA server reachable over opc.tcp://

Quick start

composer install
php bin/collector-cron.php           # one-shot
php bin/collector-daemon.php         # long-running, Ctrl+C to stop

The default configuration connects to opc.tcp://localhost:4840 and reads three standard Server status nodes (i=2257, i=2258, i=2259), which most OPC UA servers expose.

Override the endpoint without touching the file:

OPCUA_ENDPOINT=opc.tcp://192.168.1.10:4840 LOG_LEVEL=debug php bin/collector-daemon.php

---

Choosing a mode

Mode	Pattern	Network usage	Latency
Daemon	OPC UA subscription	Push from server, only on change + keep-alive	~publishingInterval ms
Cron	readMulti batch + exit	One round-trip per invocation	Whatever cron interval you set

The daemon suits continuous monitoring, alarms, and anything where you want fresh data. The cron suits periodic snapshots, hourly aggregates, audits, and environments where a long-running PHP process is impractical.

You can also run both at once: cron for an audit trail, daemon for live state. They share config.php and listeners.php.

Each cron invocation pays a fixed overhead (Composer autoload, connect, handshake, session activation). If you find yourself scheduling it more often than about once a minute, the daemon is probably the better fit.

---

Customizing the scaffold

There are two files you should normally need to touch: config.php (what to connect to) and listeners.php (what to do with the data). The rest of src/ is plumbing.

config.php

Every key is documented inline with its unit and tradeoff. The most common edits:

Key	What it controls
endpoint	OPC UA server URL — also overridable via OPCUA_ENDPOINT env var
timeout (ms)	Per-operation socket timeout
security.policy / mode	e.g. Basic256Sha256 + SignAndEncrypt for encrypted connections
security.username / password	Username/password authentication
security.certificate / privateKey / trusted	X.509 certificate authentication
subscription.publishingInterval (ms)	How often the server checks monitored items — daemon only
subscription.lifetimeCount	Grace count before the server discards the subscription
subscription.maxKeepAliveCount	Heartbeat cadence when no data changes
reconnect.backoff (ms[])	Per-attempt sleep schedule for the daemon supervisor
tags[]	What to collect: name, nodeId, optional per-tag samplingInterval

Tags accept any valid NodeId string — i=2259, ns=2;i=1001, ns=2;s=Temperature.

listeners.php

This is where your pipeline lives. The collectors load this file once at boot and pass two arguments to the closure it returns:

• App\EventDispatcher\SimpleDispatcher $bus — register listeners with $bus->on(EventClass::class, fn ($e) => …)
• Psr\Log\LoggerInterface $logger — the same logger the collectors use

Two families of events flow through $bus:

1. App\Event\TagSampled — emitted by the collectors. One per tag every time a sample is collected. Carries ->tag, ->dataValue, ->collectedAt, ->source ('daemon' or 'cron'). This is the event your storage layer listens to.

2. PhpOpcua\Client\Event\… — the library's own events: connection lifecycle, retries, alarms, raw DataChangeReceived, MonitoredItemCreated, etc. Reference:

   • offline (matches the installed version): vendor/php-opcua/opcua-client/doc/14-events.md
   • sources: vendor/php-opcua/opcua-client/src/Event/*.php

Examples

Persist samples to PostgreSQL via PDO:

$pdo = new PDO('pgsql:host=db;dbname=metrics', 'user', 'pass');
$stmt = $pdo->prepare('INSERT INTO samples(ts, tag, value, status) VALUES (?,?,?,?)');

$bus->on(TagSampled::class, function (TagSampled $e) use ($stmt): void {
    $stmt->execute([
        $e->collectedAt->format('c'),
        $e->tag->name,
        (string) $e->dataValue->getValue(),
        $e->dataValue->statusCode,
    ]);
});

Forward to MQTT (requires php-mqtt/client):

$mqtt = new \PhpMqtt\Client\MqttClient('broker.lan', 1883, 'iot-collector');
$mqtt->connect();

$bus->on(TagSampled::class, function (TagSampled $e) use ($mqtt): void {
    $mqtt->publish("opcua/{$e->tag->name}", (string) $e->dataValue->getValue(), 0, true);
});

React to an alarm:

use PhpOpcua\Client\Event\AlarmActivated;

$bus->on(AlarmActivated::class, function (AlarmActivated $e) use ($logger): void {
    $logger->warning('Alarm {source} severity {sev}', [
        'source' => $e->sourceName,
        'sev'    => $e->severity,
    ]);
    // notify your incident tooling here
});

You can register as many listeners per event class as you want — they run in registration order.

Adding a new tag

Append an entry to config.php → tags:

['name' => 'Reactor.Temperature', 'nodeId' => 'ns=2;s=Reactor.T1', 'samplingInterval' => 250],

The daemon will pick it up on the next start; for the cron, just re-run.

Adding a custom event

If you need a higher-level event — say BatchCompleted aggregating many TagSampleds — define it under src/Event/ and dispatch it from your own listener. The SimpleDispatcher matches on instanceof, so listeners registered for a base class also receive subclasses.

---

Tuning

Subscription knobs (subscription.*)

Knob	Effect of lowering it	Effect of raising it
publishingInterval	Fresher data, more PublishResponses	Less network/CPU, possible aggregation of multiple changes per cycle
samplingInterval (per tag)	Server samples the variable more often (more server load)	Server samples less often, may miss fast transients
maxKeepAliveCount	Idle heartbeats arrive sooner	Less traffic when nothing changes
lifetimeCount	Subscription expires faster on blackouts	More tolerance to client outages

For slow-changing tags (temperatures, counters), the default publishingInterval = 1000 with no per-tag samplingInterval is a reasonable starting point. For faster process variables, lower publishingInterval and set per-tag samplingInterval to match. All values are requested: the server may revise them, and the daemon logs the revised values at startup. OPC UA expects lifetimeCount >= 3 × maxKeepAliveCount.

When a server emits many changes in a burst, a per-tag queueSize in config.php asks the server to buffer the most recent N samples per item between publish cycles.

Keep listeners fast

The publish loop is sequential: publish() → dispatch all notifications → publish() again. Listener latency directly delays the next publish call. For slow work (HTTP calls, DB writes with retries, MQTT QoS > 0), consider enqueueing from the listener and draining the queue in a separate worker:

$redis = new Redis();
$redis->connect('/run/redis/redis.sock');

$bus->on(TagSampled::class, function (TagSampled $e) use ($redis): void {
    $redis->xAdd('opcua:samples', '*', [
        'ts'  => $e->collectedAt->format('U.u'),
        'tag' => $e->tag->name,
        'val' => (string) $e->dataValue->getValue(),
    ]);
});

Create connections and prepared statements once, in the closure's use(), not inside the listener body.

Listener exceptions

An exception thrown by a listener escapes the publish loop and triggers a session rebuild — a heavy recovery path for what may be a logic bug. If a listener talks to something that can hiccup, wrap its body:

$bus->on(TagSampled::class, function (TagSampled $e) use ($redis, $logger): void {
    try {
        $redis->xAdd('opcua:samples', '*', [...]);
    } catch (\Throwable $err) {
        $logger->error('listener failed: {msg}', ['msg' => $err->getMessage()]);
    }
});

Memory

The collectors do not retain sample data — they dispatch and forget. Any buffering happens in your listeners, so bound any queues or aggregations you create there. Setting a memory limit in the systemd unit (see below) is a sensible backstop.

---

Failure handling

Two retry layers

1. Library-level — opcua-client's auto-retry transparently retries individual operations on transient ConnectionException once connected. Visible at LOG_LEVEL=debug.
2. Application-level — the supervisor in DaemonCollector. When an error escapes the library, it disconnects, sleeps per reconnect.backoff, and rebuilds the session from config.php: connect, secure channel, session, subscription, monitored items. It retries until stopped with SIGINT/SIGTERM, which are honored within ~1s. Listeners don't need to handle reconnects themselves.

The reconnect schedule is configurable; the n-th failed attempt sleeps backoff[n-1] ms, and the last value repeats for subsequent attempts:

'reconnect' => [
    'backoff' => [1000, 2000, 4000, 8000, 16000, 30000],
],

What the supervisor retries

• ConnectionException — TCP / socket loss
• HandshakeException — handshake failed mid-operation
• ProtocolException — protocol-level inconsistency
• ServiceException — server-reported service error

These cover the usual cases of server restarts, failovers, and network outages. Anything else (ConfigurationException, SecurityException, InvalidArgumentException, …) exits with code 1: configuration mistakes should fail loudly, not be retried.

Liveness

The library dispatches SubscriptionKeepAlive whenever the server heartbeats with no notifications, so a simple liveness check is "when did I last see either a TagSampled or a SubscriptionKeepAlive":

use PhpOpcua\Client\Event\SubscriptionKeepAlive;

$lastSignal = time();
$bus->on(TagSampled::class,            function () use (&$lastSignal) { $lastSignal = time(); });
$bus->on(SubscriptionKeepAlive::class, function () use (&$lastSignal) { $lastSignal = time(); });

If that timestamp grows older than a few multiples of publishingInterval × maxKeepAliveCount, the daemon is stalled.

Cron exit codes

Code	Meaning
0	Success — every tag returned a Good status
2	Connected and read, but at least one tag had a non-Good status
1	Fatal error — connect failed, config invalid, exception escaped

Exit 2 is a data-quality failure rather than an infrastructure failure — useful to distinguish in alerting.

---

Use cases

• Continuous monitoring dashboards — daemon writes TagSampled to a time-series database, with a dashboard on top.
• Feature pipelines — daemon publishes TagSampled to a message broker; downstream consumers do the processing.
• Alarm forwarding — daemon listens for AlarmActivated and notifies your incident tooling.
• Periodic audits — cron runs hourly and dumps selected process variables to a file for long-term storage.
• Quick exploration — for one-off reads and browsing, the opcua-cli tool is simpler than this scaffold.

What this scaffold is not for:

• Bidirectional control (writing setpoints back to PLCs) — possible with the underlying client ($client->write(...)) but intentionally not wired up here. Writing back into a live process from an event listener needs careful gating; design that path deliberately.
• Multi-server fan-in — this scaffold connects to one endpoint. For N servers, run N instances (per-server config) and federate downstream.
• Hard real-time requirements — PHP is not a real-time runtime.

---

Running in production

systemd unit

# /etc/systemd/system/opcua-collector.service
[Unit]
Description=OPC UA IoT data collector
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=opcua
WorkingDirectory=/opt/opcua-collector
ExecStart=/usr/bin/php /opt/opcua-collector/bin/collector-daemon.php
Restart=always
RestartSec=5
MemoryMax=512M
Environment=LOG_LEVEL=info

[Install]
WantedBy=multi-user.target

systemctl enable --now opcua-collector
journalctl -u opcua-collector -f

The daemon exits 0 on SIGTERM (graceful), so systemd's Restart=always only kicks in on real crashes. Reconnects are handled internally — systemd won't see them. MemoryMax is a backstop: if a listener ever leaks, the process is restarted cleanly instead of growing without bound.

cron entry

*/5 * * * * cd /opt/opcua-collector && /usr/bin/php bin/collector-cron.php >> /var/log/opcua-cron.log 2>&1

---

Testing

# Lint everything
find src bin config.php listeners.php -name '*.php' -exec php -l {} \;

# Smoke test against a reachable server
php bin/collector-cron.php

# Smoke test the supervisor with an unreachable endpoint
timeout 12 env OPCUA_ENDPOINT=opc.tcp://127.0.0.1:9999 php bin/collector-daemon.php

Worth verifying in a staging environment before relying on the daemon: that it reconnects after a network interruption and after a server restart, and that a deliberately slow or throwing listener behaves the way you expect.

For unit-testing your listeners without a real server, see vendor/php-opcua/opcua-client/doc/13-testing.md and use MockClient.

---

License

MIT.