Reference — Messaging Interface

The topic/message contract: what the processor subscribes to, what it publishes, and where. For configuration of routes and targets see configuration.md; for value typing see data-types.md; for the design rationale see explanation.md.

The processor is a transform-and-forward stage on the edgecommons Unified Namespace (UNS): it subscribes to the fleet’s telemetry (the data class), runs a per-route pipeline, and forwards each result to one target. It is also — for free from the edgecommons library — a first-class command citizen: it answers the built-in cmd verbs and its own custom verbs (see Command verbs), publishes evt health events, and emits a metric throughput metric.

Unified Namespace (UNS) topic grammar

Every UNS topic is ecv1/{device}/{component}/{instance}/{class}[/channel] (rootless; a site position appears between ecv1 and {device} only under a multi-level hierarchy with topic.includeRoot). The processor’s own token is telemetry-processor (the sanitized short name after the last . of com.mbreissi.edgecommons.TelemetryProcessor), so it lives at ecv1/{device}/telemetry-processor/{instance}/{class}[/channel].

The eight classes, and how the processor uses each:

Class	Owner	Processor use
data	application	Input (subscribes the fleet’s adapter data) and output (processed telemetry)
evt	application	Output — pipeline health events (see Events) and forwarded alarms
cmd	application	Input — the library command inbox (built-in + custom verbs)
app	application	free-form; unused by default
state	reserved (library)	the automatic keepalive on ecv1/{device}/telemetry-processor/state
metric	reserved (library)	the processor’s metric/pipeline throughput metric
cfg	reserved (library)	the effective-config publisher
log	reserved (library)	(library-owned)

Reserved-class publish guard. A component publish (publish / publish_northbound) to a state / metric / cfg / log topic is rejected by the library — the reserved classes are written only through the library’s own publishers. A route publish.topic must therefore target data / evt / app; the processor logs a WARN at startup if a resolved publish.topic lands on a reserved class (the publish would otherwise be silently dropped). Subscribing a reserved class to read it is always allowed.

Envelope

Normal messages arrive on the wire as EdgeCommons protobuf envelopes. The processor exposes the same diagnostic/projection shape to filters, scripts, key paths, and file rows — { header, identity, tags, body } — that the adapters publish:

{
  "header":   { "name": "SouthboundSignalUpdate", "version": "1.0", "timestamp": "<ISO-8601>",
                "uuid": "…", "correlation_id": null },
  "identity": { "hier": [ { "level": "device", "value": "<device>" } ],
                "path": "<device>", "component": "opcua-adapter", "instance": "kep1" },
  "tags":     { "appId": "…", "site": "…", "shop": "…", "line": "…" },
  "body": {
    "device":  { "adapter": "opcua", "instance": "<instanceId>", "endpoint": "opc.tcp://host:4840" },
    "signal":  { "id": "<canonical stable id>", "name": "<human label>", "address": { /* protocol-native */ } },
    "samples": [
      { "value": <any>, "quality": "GOOD|BAD|UNCERTAIN", "qualityRaw": "<native code>",
        "sourceTs": "<ISO-8601 UTC>", "serverTs": "<ISO-8601 UTC>" }
    ]
  }
}

The processor does not require a specific header.name: any JSON message that matches a route’s subscribe filter flows through (filters and scripts can act on any body).

Identity vs. envelope tags vs. the signal — three different things

• The identity element — the top-level UNS identity of the publisher: hier (the hierarchy levels, the last of which is the device), the precomputed path, the component token, and the per-message instance. This is where the source device lives. Pipelines read it via the identity. JSON path (identity.device / identity.component / identity.instance / identity.path) and scripts read the identity binding. This is the correct key for “which device/adapter produced this reading”.
• Envelope tags — the edgecommons message-envelope metadata map: an open set of key/values that ride on every message (appId, site, shop, line, or any custom key). Opaque business metadata: exposed to scripts as the tags binding, usable in topic templates, and landed by the file sink’s default projection in one JSON column. (A thing key in tags is just an ordinary tag with no special meaning.)
• The signal — one southbound data point (an OPC UA node, a Modbus register, …) carried in body.signal ({ id, name, address }) with its readings in body.samples[]. Historically called a “tag” in the OPC UA / historian world; the edgecommons contract calls it a signal.

The identity. JSON path

Filter field, aggregate/sample by, the route key, and stream partitionKey all take a dotted key path. Alongside body. / tags. / header. there is also an identity. root exposing the source publisher’s UNS identity:

Path	Value
identity.device	the source device (last hierarchy value)
identity.component	the source component token (e.g. opcua-adapter)
identity.instance	the source per-message instance token
identity.path	the /-joined hierarchy values
identity.hier[].level / identity.hier[].value	the hierarchy levels (array spread)

Scripts get the same view as an identity scope binding (identity.device, identity.component, …).

Subscribes

Each route subscribes to its configured subscribe filters on the local bus (MQTT/IPC). MQTT wildcards (+, #) are allowed; {ThingName} / {ComponentName} / tags{} template variables are substituted at startup (see configuration.md). A filter shared by several routes is subscribed once and fanned out to each route.

The fleet consumer for all southbound telemetry is the single UNS wildcard:

ecv1/+/+/+/data/#          # every device / component / instance / signal on the data class

Scope it to specific adapters when you don’t want the whole fleet, e.g. ecv1/+/opcua-adapter/+/data/#. A typical input is a SouthboundSignalUpdate envelope (docs/SOUTHBOUND.md §7) published by an adapter on ecv1/{device}/{adapter}/{inst}/data/{signalPath} — but the processor is payload-agnostic (below).

The processor is payload-agnostic — it does not mandate this schema. Any JSON body that matches a route’s subscribe filter flows through; the SouthboundSignalUpdate shape is only a convention the built-in conveniences default to. Specifically, the shape is assumed only by: the quality filter shorthand, the default key body.signal.id, body.samples[] aggregation, and the file sink’s default rows projection. Each is overridable — point a filter field/script at your own paths, set the route key and the aggregate value path, and declare a rows user projection.

Self-echo guard (loop safety)

Because the processor consumes the data class it also republishes onto (for local targets), a naive fleet subscription would re-consume its own output → an amplifying loop. Two mechanisms prevent this:

1. Restamp on local. For a local target the dispatcher restamps the output envelope’s identity with the processor’s own identity (instance = the route id). The local output is the processor’s product, and this makes step 2 effective. (northbound / stream targets leave the source identity intact — they never re-enter the local bus, so provenance is preserved.)
2. Drop own echoes. The subscribe fan-out drops any inbound message whose identity device and component equal the processor’s own. A re-consumed local output (now carrying the processor’s identity) is discarded. Cross-device chaining still works — a different device does not match.

Publishes

The output target is per route (target). Route outputs must land on a non-reserved class (data / evt / app).

target	Destination	Topic / key	Transport call
local	local bus	publish.topic (default = the source topic); identity restamped to the processor	publish(topic, msg)
northbound	AWS IoT Core / northbound MQTT	publish.topic (default = the source topic), QoS from publish.qos	publish_northbound(topic, msg, qos)
stream:<name>	a edgecommons durable stream	partition key from publish.partitionKey (default = the route key, i.e. body.signal.id)	streams().stream(name).append(record)

• Set an explicit publish.topic to a UNS data/evt/app topic template, e.g. ecv1/{ThingName}/telemetry-processor/data/downsampled or ecv1/{ThingName}/telemetry-processor/evt/alarms. Templates are resolved at startup.
• A multi-signal script stage with an output.topic carries its own topic: the derived message reaches the target on that topic, and combining it with a route-level publish.topic is a startup error (the route topic would override the stage’s).
• northbound publishes to IoT Core via the mqttproxy with qos = atLeastOnce (default) or atMostOnce.
• stream:<name> appends the EdgeCommons protobuf envelope as one record; the stream’s configured sink (kinesis/kafka/file) delivers it asynchronously. Kinesis/Kafka sinks default to a JSON projection and can opt into protobuf bytes with sink.payloadFormat: "protobuf". Forwarding errors are logged and tallied (a stream-unavailable evt fires), never propagated — use a durable stream: target for no-loss output.

Why this dispatch uses messaging()/streams() rather than the data() facade. The edgecommons library’s data() publish facade (gg.instance(id).data()) is the right tool for a southbound adapter minting a fresh SouthboundSignalUpdate from a protocol read — it owns topic minting (always under its own bound identity), the SouthboundSignalUpdate header, and quality/timestamp defaulting. This dispatcher instead republishes an already-built message that may not be southbound-shaped at all (the processor is payload-agnostic, above) and — for local/northbound — defaults to the exact source topic (a deliberate bridging behavior the alarms-northbound sample relies on). Routing that through data() would change the published topic and force a header/body shape the processor does not require, so the dispatcher uses the lower-level path. The route target itself is the library’s own edgecommons::facades::Channel (local | northbound | stream:<name>) rather than a bespoke processor type, so the routing vocabulary is shared even though the dispatch mechanics are not. evt health events (above) use the matching events() facade, where the facade’s identity/topic/body ownership matches exactly what the processor needs.

Aggregate output (ProcessedTelemetry)

An aggregate stage emits one message per (key, window) with header.name = "ProcessedTelemetry" (other envelope fields inherited from the window’s first message — including the source identity, except on a local target where it is restamped):

"body": {
  "signal": { "id": "<key>", … },           // the source signal identity, where present
  "samples": [ { "value": <primary>, "quality": "GOOD" } ],
  "agg": { "avg": 20.0, "max": 30.0, "min": 10.0, "count": 3, "last": 30.0 },
  "window": { "startMs": 1719446400000, "endMs": 1719446405000, "count": 3 }
}

Field	Meaning
samples[0].value	the primary reducer = the first-listed fn. Carried in samples[] so rows-mode file archiving lands a value in the typed value column.
samples[0].quality	always "GOOD".
agg.<fn>	the full reducer set (one entry per configured fn). See data-types.md for value types.
window.startMs / window.endMs	window bounds (Unix ms). For a count window both equal the close time.
window.count	number of folded sample values in the window (each body.samples[].value; an array value is folded element-wise, so this equals the message count only when each message carries a single scalar sample).

Numeric reducers (avg/max/min/sum) are emitted only when ≥1 sample in the window was numeric; otherwise that reducer is null.

Script output (ScriptResult)

A multi-signal script stage with a configured output publishes one new envelope per successful evaluation on output.topic (configuration reference):

Envelope field	Value
header.name / header.version	output.name / output.version — default ScriptResult / 1.0.
header.uuid / header.timestamp	fresh, minted for the derived message.
header.correlation_id	the triggering message’s uuid — trace a result back to the exact update that fired it.
identity	the processor’s own identity with instance = the route id — the derived signal’s producer is the processor, on every target (not only local).
body	exactly what the script returned (map/table → JSON object).

The triggering input message is consumed, never republished on the output topic. Without a configured output, a script stage keeps its in-place contract: the result replaces the body of the triggering message, which continues on the source topic.

Command verbs

The processor subscribes its own command inbox ecv1/{device}/telemetry-processor/cmd/# and, for each configured route, the per-route inbox ecv1/{device}/telemetry-processor/{route}/cmd/# (both wired automatically by the library). A cmd request whose header.reply_to is set gets a structured reply {"ok": true, "result": …} or {"ok": false, "error": {"code", "message"}}; a request without reply_to is fire-and-forget.

Built-in verbs (library-provided, cannot be shadowed):

Verb	Scope	Result
ping	both	{ "status": "RUNNING", "uptimeSecs": n } — liveness/echo
reload-config	both	re-fetch + re-apply the config from the active source → { "reloaded": true }
get-configuration	both	the current redacted effective config → { "config": … }

Custom verbs (registered by the processor):

Verb	Scope	Body	Result
get-stats	both	—	{ "routes": [ { id, in, out, dropped, streamAppends, publishFailures, queueDepth, paused } ] } — counters for the addressed route, or every route
flush	both	—	force-close the addressed route’s (or every route’s) open time windows now → { "flushed": n } (messages emitted). Count windows keep their count semantics.
pause	both	{ "route"? }	stop enqueuing to a route → { "paused": [ids] }
resume	both	{ "route"? }	the inverse of pause → { "resumed": [ids] }

Verb scope and addressing

A route is a component.instances[] entry, so every verb above is addressable two ways and each one declares the scope both:

• Component-addressed (ecv1/{device}/telemetry-processor/cmd/{verb}, no route token) means every route — the fleet-wide form.
• Route-addressed (ecv1/{device}/telemetry-processor/{route}/cmd/{verb}) acts on that route alone.

The topic’s route token is authoritative: when a request is route-addressed, pause/resume ignore a route field in the body. The route body field remains the way to target a single route over the component-addressed topic. A request that names one route in the topic and a different instance in body.instance is rejected by the library with BAD_ARGS before the verb runs.

Known limitation. The built-in reload-config hot-swaps the config snapshot but the routes are wired once at startup, so a route topology change needs a component restart; there is no dynamic route rebuild. pause/resume and flush operate on the already-wired routes.

Events (evt)

The processor publishes rate-limited health events through the edgecommons events() publish facade (gg.events(), at component scope — no instance token) on ecv1/{device}/telemetry-processor/evt/{severity}/{type} (a non-reserved class; subscribe the fleet with ecv1/+/+/+/evt/#). The facade derives the channel from the body’s own severity/type (so topic and body can never disagree) and stamps timestamp; every one of the processor’s own health events currently uses severity: "warning":

type	When	Body
queue-overflow	a route’s worker queue was full and a message was dropped (backpressure)	{ "severity": "warning", "type": "queue-overflow", "timestamp", "context": { "route" } }
route-error	a local/northbound forward failed	{ "severity": "warning", "type": "route-error", "message": "<error>", "timestamp", "context": { "route", "topic" } }
stream-unavailable	a stream:<name> target is down / its append failed	{ "severity": "warning", "type": "stream-unavailable", "message": "<error>", "timestamp", "context": { "route", "stream" } }

Each type is coalesced (at most one event per type per cooldown window) so a sustained fault can’t storm the bus. Routes 2 (alarms-northbound in the recipe) can also forward alarm-flagged readings northbound as evt/alarms — that path is a plain publish.topic override on the route’s raw local/northbound dispatch (see Publishes above), not the events() facade, since it is forwarding an already-shaped signal reading, not minting a processor health event.

Metrics (metric)

With metricEmission.target: "messaging" the processor emits a pipeline metric every 30 s on ecv1/{device}/telemetry-processor/metric/pipeline (subscribe ecv1/+/+/+/metric/#), carrying the summed per-interval deltas of the route counters: messagesIn, messagesOut, messagesDropped, streamAppends, publishFailures. Per-route detail is available on demand via the get-stats command. System measures (CPU/memory/…) are emitted automatically by the heartbeat as the sys metric.

For the purpose of every processor metric and what each measure helps determine, see Reference - Metrics.

Topic-template variables

Resolved (once, at startup) into subscribe filters and publish.topic:

Variable	Resolves to
{ThingName}	the -t/--thing value (or platform identity) — the device
{ComponentName} / {ComponentFullName}	the component’s short (telemetry-processor) / fully-qualified name
{<key>}	any key under top-level tags — {site}, {appId}, {shop}, {line}, or any custom key

publish.partitionKey is not a topic template — it is a key path resolved against each message (default body.signal.id).

Channel guidance

Route each result to the right destination:

• High-rate / bulk telemetry → stream:<name> (durable buffer → Kinesis/Kafka/file). The channel for the firehose; the archive preserves the source identity for provenance.
• Low-rate control/alarm data → northbound (IoT Core, e.g. an evt/alarms topic) or local (re-publish on the bus as the data class for another local consumer).

Sizing a route’s maxQueue generously and preferring a stream: target gives the strongest delivery guarantee; local/northbound dispatch is fire-and-forget (failures are tallied + surfaced as evt).