Messaging

The messaging subsystem gives your component one interface over two transports: Greengrass IPC (--transport IPC) on a Nucleus-managed device, or a dual-MQTT provider (--transport MQTT) that connects to a local broker and, optionally, AWS IoT Core. The same publish/subscribe, request/reply, and message envelope work identically across Java, Python, Rust, and TypeScript, so components written in any language interoperate on the same topics.

Topics live in the Unified Namespace: mint them with the gg.uns() builder rather than concatenating strings, and every envelope built through a config-bound builder carries a self-identifying identity element.

Connections and subscriptions are synchronous — connect, publish, and subscribe run inline rather than returning a pending promise. Be aware, though, that the standalone (dual-MQTT) provider swallows the underlying Paho MqttException (it logs the error and returns), so a normal return does not by itself guarantee the broker accepted the publish/subscribe — check the logs (and connected()) when diagnosing a missing message.

Reaching the messaging service

Every component reaches messaging through its EdgeCommons handle. The accessor name and return shape differ slightly per language.

Java

getMessaging() returns a MessagingClient instance — its methods are instance methods.

MessagingClient messaging = gg.getMessaging();

Python

get_messaging() returns the MessagingClient class — its methods are @staticmethod, so you call them on the returned class.

messaging = gg.get_messaging()

Rust

messaging() returns a Result<Arc<dyn MessagingService>> — it errors if no transport was wired.

let messaging = gg.messaging()?;   // Arc<dyn MessagingService>

TypeScript

messaging() returns an IMessagingService — it throws if no transport was wired.

const messaging = gg.messaging();   // IMessagingService

The message envelope

Every message EdgeCommons sends on the wire is a JSON object with four parts. The shape and key names are identical across all four languages so they interoperate.

{
  "header": {
    "name": "TemperatureReading",
    "version": "1.0",
    "timestamp": "2026-06-29T12:00:00Z",
    "uuid": "5f6c...",
    "correlation_id": "a1b2...",
    "reply_to": "edgecommons/reply-..."
  },
  "identity": {
    "hier": [
      { "level": "site",   "value": "dallas" },
      { "level": "device", "value": "gw-01" }
    ],
    "path": "dallas/gw-01",
    "component": "my-component",
    "instance": "main"
  },
  "tags": { "site": "lab" },
  "body": { "celsius": 22.4 }
}

• header — message metadata. Keys are snake_case on the wire (correlation_id, reply_to, uuid). reply_to is omitted when there is no reply topic; correlation_id/uuid/timestamp are filled in automatically if you do not set them.
• identity — the publisher’s UNS identity: the ordered enterprise hierarchy (hier, last entry = the device), the precomputed path, and the component/instance addressing tokens. Stamped automatically by any config-bound builder; a message built without config (a bootstrap request, raw bridging) legally omits it. Consumers identify the sender from this element — never by parsing the topic.
• tags — business context only (app, org, cost center) — your custom tags config keys. The device identity lives in the identity element, not in tags. tags is emitted only when tags were stamped.
• body — your payload. Any JSON value.

For small binary payloads, the body uses an explicit marker object rather than a bare base64 string:

{
  "_edgecommonsBinary": {
    "encoding": "base64",
    "length": 5,
    "data": "AAEC/v8="
  }
}

This is intentionally bounded control-payload support, not a frame/video streaming transport: decoded binary bodies are limited to 64 KiB. Java byte[], Python bytes/bytearray, and TypeScript Buffer/Uint8Array payloads serialize to this marker. Rust uses MessageBuilder::binary_payload(...). Consumers keep using getBody() / get_body() / getBody() / body for the raw JSON marker, or use the typed binary accessors getBinaryBody() / get_binary_body() / binary_body() / getBinaryBody() to decode and validate it.

A raw message has no envelope — it serializes as { "raw": <value> } instead. When a component receives a payload that is not an envelope (none of header/identity/tags/body), or non-JSON bytes, it is delivered to your handler as a raw message (non-JSON bytes become a raw string). See Raw messages below.

Note

The reply-topic prefix is the literal string edgecommons/reply- (a trailing -, not a /) in all four languages.

Building a message

Construct messages with MessageBuilder — not the raw constructors. The builder stamps the uuid, correlation_id, and timestamp for you, and — when you supply the config — the custom tags and the UNS identity element (with the optional per-message instance token — omit it for a component-scoped message).

Note

build() never requires config in any language. Supplying the config is what stamps the custom tags and the config-resolved identity on the envelope; a builder without config produces a legal envelope with no identity (the bootstrap/raw-bridging case). Use withInstance(...) / with_instance(...) / .instance(...) to stamp an instance token — or better, build through an instance handle.

Java

Message msg = MessageBuilder.create("TemperatureReading", "1.0")
    .withPayload(payload)        // JsonObject, Map, POJO, or List
    .withConfig(configManager)   // stamps tags + the UNS identity
    .withInstance("kep1")        // optional per-message instance token (omit for component scope)
    .build();

Python

msg = (
    MessageBuilder.create("TemperatureReading", "1.0")
    .with_payload({"celsius": 22.4})
    .with_config(config)         # stamps tags + the UNS identity
    .with_instance("kep1")       # optional per-message instance token (omit for component scope)
    .build()
)

Rust

Rust uses different method names: new (not create), payload (not with_payload), and from_config (not with_config).

use serde_json::json;

let msg = MessageBuilder::new("TemperatureReading", "1.0")
    .payload(json!({ "celsius": 22.4 }))
    .from_config(&config)        // stamps tags + the UNS identity
    .instance("kep1")            // optional per-message instance token (omit for component scope)
    .build();

TypeScript

const msg = MessageBuilder.create("TemperatureReading", "1.0")
  .withPayload({ celsius: 22.4 })
  .withConfig(gg.config())       // stamps tags + the UNS identity
  .withInstance("kep1")          // optional per-message instance token (omit for component scope)
  .build();

For tests, vectors, and relays, an explicit withIdentity(...) / with_identity(...) / .identity(...) override always wins over the config-resolved identity. All four builders also accept explicit uuid and timestamp setters for deterministic envelopes.

UNS topics and the reserved-class guard

messaging() takes literal topics — there is no hidden topic construction. For a topic in the Unified Namespace, mint it with gg.uns() (or an instance handle’s uns()), which validates the token charset and the ≤ 7-slash IoT Core depth budget at build time:

Java

String topic = gg.getUns().topic(UnsClass.APP, "order/received");
gg.getMessaging().publish(topic, msg);   // ecv1/{device}/{component}/app/order/received

Python

from edgecommons.uns import UnsClass

topic = gg.uns().topic(UnsClass.APP, "order/received")
gg.get_messaging().publish(topic, msg)   # ecv1/{device}/{component}/app/order/received

Rust

use edgecommons::uns::UnsClass;

let topic = gg.uns().topic_with_channel(UnsClass::App, "order/received")?;
gg.messaging()?.publish(&topic, &msg).await?;

TypeScript

import { UnsClass } from "@edgecommons/edgecommons";

const topic = gg.uns().topic(UnsClass.App, "order/received");
await gg.messaging().publish(topic, msg);

Arbitrary non-UNS topics remain fully supported (bridging an external MQTT system, the cloudwatch/metric/put contract). The single restriction is the reserved-class guard: a raw publish (or request/reply) to a reserved-class UNS topic — ecv1/.../{state|metric|cfg|log} — is rejected (ReservedTopicException / ReservedTopicError / EdgeCommonsError::ReservedTopic), because those classes are library-owned: the heartbeat publishes state, the metrics subsystem publishes metric, and the config subsystem publishes cfg. Subscribing to reserved classes is always allowed. See the UNS API reference.

Publish and subscribe

publish sends an envelope to a topic. subscribe registers a handler that receives the topic and the decoded Message for each match. Topic filters use the usual MQTT wildcards (+ single level, # multi level). The examples below use a literal topic; UNS-native code mints the topic via uns() as shown above.

Danger

The subscribe integer arguments differ in order AND meaning across languages — read this before copying a snippet.

• Java / Python: subscribe(topic, callback, maxConcurrency[, maxMessages]) — the first int is maxConcurrency.
• Rust / TypeScript: subscribe(filter, handler, maxMessages, maxConcurrency) — the first int is maxMessages.

So a trailing , 3 means 3 concurrent handlers in Java/Python but a queue depth of 3 in TypeScript. See Per-subscription concurrency caps.

Java

// subscribe — 3rd arg is maxConcurrency. The callback is a BiConsumer<String,Message>.
messaging.subscribe("sensors/temp", (topic, m) -> {
    System.out.println(m.getHeader().getName() + " -> " + m.getBody());
}, 1);

// publish
messaging.publish("sensors/temp", msg);

Python

def handler(topic, m):
    print(m.get_header().name, "->", m.get_body())

# subscribe — 3rd arg is max_concurrency
messaging.subscribe("sensors/temp", handler, 1)

# publish
messaging.publish("sensors/temp", msg)

Rust

The handler closure must be wrapped with message_handler(...) — a bare closure is not accepted. The two trailing ints are (max_messages, max_concurrency).

use edgecommons::messaging::message_handler;

messaging.subscribe(
    "sensors/temp",
    message_handler(|topic, msg| async move {
        tracing::info!(%topic, name = %msg.header.name, "received");
    }),
    32,  // max_messages
    1,   // max_concurrency
).await?;

messaging.publish("sensors/temp", &msg).await?;

TypeScript

The two trailing ints are (maxMessages, maxConcurrency); they default to 32 and 1.

await messaging.subscribe("sensors/temp", (topic, m) => {
  console.log(m.header.name, "->", m.getBody());
}, 32, 1);

await messaging.publish("sensors/temp", msg);

Raw publish and subscribe

Use the *Raw publish methods to send an unwrapped JSON value (for example to interoperate with a non-EdgeCommons producer). On the receiving side, you do not call a separate “subscribe raw” — your normal subscribe handler receives a raw Message whenever the incoming payload is not an envelope, and you detect it with the raw accessors.

Java

// publish a raw JSON object (no envelope)
JsonObject metric = new JsonObject();
metric.addProperty("value", 42);
messaging.publishRaw("metrics/raw", metric);

// handle a possibly-raw message
messaging.subscribe("metrics/raw", (topic, m) -> {
    if (m.getRaw() != null) {
        System.out.println("raw: " + m.getRaw());
    } else {
        System.out.println("envelope body: " + m.getBody());
    }
}, 1);

Python

# publish a raw dict (no envelope)
messaging.publish_raw("metrics/raw", {"value": 42})

# handle a possibly-raw message
def handler(topic, m):
    if m.get_raw() is not None:
        print("raw:", m.get_raw())
    else:
        print("envelope body:", m.get_body())

messaging.subscribe("metrics/raw", handler, 1)

Rust

use serde_json::json;

// publish a raw JSON value (no envelope)
messaging.publish_raw("metrics/raw", &json!({ "value": 42 })).await?;

// handle a possibly-raw message
messaging.subscribe(
    "metrics/raw",
    message_handler(|_topic, msg| async move {
        match msg.get_raw() {
            Some(v) => tracing::info!(?v, "raw"),
            None => tracing::info!(body = ?msg.body, "envelope"),
        }
    }),
    32,
    1,
).await?;

TypeScript

// publish a raw value (no envelope)
await messaging.publishRaw("metrics/raw", { value: 42 });

// handle a possibly-raw message
await messaging.subscribe("metrics/raw", (_topic, m) => {
  if (m.isRaw()) {
    console.log("raw:", m.getRaw());
  } else {
    console.log("envelope body:", m.getBody());
  }
}, 32, 1);

Request/reply with correlation

request publishes a message to a service topic, subscribes to a private reply topic, and gives you a future that completes when the reply arrives. On the responder side, reply(request, reply) copies the request’s correlation_id onto the reply and publishes it to the request’s reply_to topic — so the requester’s future resolves to the right answer.

Every request carries a framework-owned deadline. The default is messaging.requestTimeoutSeconds (30 s; 0 disables); an explicit per-call timeout always wins. When the deadline fires, the library unsubscribes the ephemeral reply topic and completes the future exceptionally — even if you never await it — so an abandoned request does not leak its reply subscription (and trip the Greengrass shared-connection quota). Consequences: a no-arg get()/await is bounded by the deadline, and a get(t) waits at most min(t, deadline).

The future type differs per language: Java extends CompletableFuture<Message>, Python returns an Iou, Rust returns a Result<ReplyFuture> you await, and TypeScript returns a ReplyFuture (PromiseLike) you await.

Java

ReplyFuture extends CompletableFuture<Message>, so use the standard future API. On deadline it completes exceptionally with a java.util.concurrent.TimeoutException. The per-call override is request(topic, msg, Duration) (null = the default, Duration.ZERO = disabled).

ReplyFuture future = messaging.request("svc/op", request);          // 30 s default deadline
Message reply = future.get();                                        // safe: bounded by the deadline

ReplyFuture slow = messaging.request("svc/op", request, Duration.ofSeconds(120));

// responder side:
messaging.reply(request, replyMsg);   // copies correlation id automatically

Python

request(topic, msg, timeout_secs=None) returns an Iou (None = the configured default, 0 = disabled). Its get(timeout) returns a (done, result) tuple — but when the framework deadline fired first, get() raises RequestTimeoutError.

from edgecommons.messaging.errors import RequestTimeoutError

iou = messaging.request("svc/op", request)          # 30 s default deadline
try:
    done, reply = iou.get(timeout=5)                # (done, result) tuple
    if done:
        print(reply.get_body())
except RequestTimeoutError:
    ...                                              # the framework deadline fired

# responder side:
messaging.reply(request, reply_msg)     # copies correlation id automatically

Rust

request() returns Result<ReplyFuture> — await the call to get the future, then await the future. On deadline it resolves Err(EdgeCommonsError::RequestTimeout { .. }). The per-call override is request_with_timeout(topic, msg, Option<Duration>) (None = the default, Some(Duration::ZERO) = disabled). Dropping the ReplyFuture still cancels the request.

use std::time::Duration;

let fut = messaging.request("svc/op", request).await?;   // 30 s default deadline
let reply = fut.await?;                                    // Err(EdgeCommonsError::RequestTimeout) on deadline

let fut = messaging
    .request_with_timeout("svc/op", request, Some(Duration::from_secs(120)))
    .await?;

// responder side:
responder.reply(&request, reply_msg).await?;   // copies correlation id automatically

TypeScript

request(topic, msg, timeoutMs?) returns a ReplyFuture you await. An omitted timeoutMs uses the configured default (30 s); an explicit 0 disables the deadline. On deadline the promise rejects with RequestTimeoutError.

const reply = await messaging.request("svc/op", req);        // 30 s default deadline
console.log(reply.getBody());

const slow = await messaging.request("svc/op", req, 120_000); // explicit per-call deadline

// responder side:
await messaging.reply(req, replyMsg);   // copies correlation id automatically

To abandon an in-flight request before it completes, call cancelRequest / cancel_request (Rust/TS also expose .cancel() on the future), which tears down the reply subscription. Reply-arrival, the deadline, and cancellation race through one idempotent settle path — exactly one wins; a straggler reply after settle is logged at DEBUG and dropped.

Per-subscription concurrency caps

Each subscription has two independent bounds:

• maxConcurrency — the maximum number of handler invocations that run at once for that subscription. A value <= 0 means uncapped (Java/Python skip the semaphore; Rust/TS clamp the value up to 1). Use a low cap to serialize a handler that touches shared state; raise it to process bursts in parallel.
• maxMessages — the bounded in-memory queue depth per subscription. When the queue is full, new messages are dropped with a warning (the subscription is never blocked). The default is 10000 in Java/Python (DEFAULT_MAX_MESSAGES), 32 in TypeScript, and is caller-supplied in Rust.

Remember the argument-order divergence (see the warning above): Java/Python take maxConcurrency first, Rust/TypeScript take maxMessages first.

Destination: local vs northbound, and QoS

With the dual-MQTT transport, the plain methods (publish, subscribe, request, reply) target the local broker. Northbound methods target the configured northbound transport and take a QoS:

Local	Northbound
publish / publishRaw	publishNorthbound / publishNorthboundRaw
subscribe	subscribeNorthbound
request	requestNorthbound
reply	replyNorthbound
unsubscribe	unsubscribeNorthbound

Note

The northbound family is named consistently in each SDK: Java and TypeScript use publishNorthbound / subscribeNorthbound, Rust uses publish_northbound plus Destination::Northbound, and Python uses snake_case (publish_northbound). Config uses northbound for the optional second broker and related route selectors.

QoS defaults are config-backed for operations that do not carry an explicit QoS parameter. The standalone local and northbound MQTT brokers support QoS 0 (at most once), 1 (at least once), and 2 (exactly once). Greengrass IoT Core IPC APIs support only QoS 0 and 1 through their explicit SDK enums. Direct northbound methods that take a QoS argument still use that per-call value.

All four languages expose an EdgeCommons-owned Qos enum for messaging APIs. The config file uses numeric MQTT QoS values:

{
  "messaging": {
    "local": {
      "host": "localhost",
      "port": 1883,
      "clientId": "my-component-local",
      "qos": { "publish": 1, "subscribe": 1 }
    },
    "northbound": {
      "host": "northbound-broker.example.com",
      "port": 8883,
      "clientId": "my-component-northbound",
      "qos": { "publish": 2, "subscribe": 1 }
    }
  }
}

messaging.local.qos.publish drives local publish, publishRaw, request publish, and reply publish. messaging.local.qos.subscribe drives local subscribe and request reply subscriptions. The matching messaging.northbound.qos defaults drive northbound MQTT request/reply operations that do not expose an explicit QoS argument.

Java

import com.mbreissi.edgecommons.messaging.Qos;

messaging.publishNorthbound("cloud/telemetry", msg, Qos.AT_LEAST_ONCE);
messaging.subscribeNorthbound("cloud/cmd", (topic, m) -> handle(m), Qos.AT_LEAST_ONCE, 1);

Python

from edgecommons.messaging.qos import Qos

messaging.publish_northbound("cloud/telemetry", msg, Qos.AT_LEAST_ONCE)
messaging.subscribe_northbound("cloud/cmd", handler, Qos.AT_LEAST_ONCE, 1)

Rust

use edgecommons::messaging::Qos;

messaging.publish_northbound("cloud/telemetry", &msg, Qos::AtLeastOnce).await?;
messaging.subscribe_northbound(
    "cloud/cmd",
    message_handler(|_t, m| async move { handle(m).await }),
    Qos::AtLeastOnce,
    32,
    1,
).await?;

TypeScript

import { Qos } from "@edgecommons/edgecommons";

await messaging.publishNorthbound("cloud/telemetry", msg, Qos.AtLeastOnce);
// subscribeNorthbound's qos arg defaults to Qos.AtLeastOnce
await messaging.subscribeNorthbound("cloud/cmd", (_t, m) => handle(m), Qos.AtLeastOnce, 32, 1);

Tip

connected() reports the local broker link only — a northbound outage does not flip it. It backs the /readyz health check (never /livez), so a component stays “ready” through transient cloud disconnects.

The dual-MQTT standalone config

On the HOST platform (and for local testing) the MQTT transport reads a JSON messaging section that describes the brokers. It is loaded from the positional argument to --transport MQTT <messaging_config.json> (on KUBERNETES, from the mounted ConfigMap default) — not from the active config source (-c); an embedded messaging section in a -c source is not read by the MQTT transport.

This separate broker file is parsed leniently (plain Gson / the equivalent JSON decoder in each language) and is not validated against the config JSON Schema at runtime — unknown or mistyped keys are silently ignored rather than rejected, so double-check the field names below.

{
  "messaging": {
    "local": {
      "host": "localhost",
      "port": 8883,
      "clientId": "my-component",
      "qos": { "publish": 1, "subscribe": 1 },
      "credentials": {
        "caPath": "/certs/ca.crt",
        "certPath": "/certs/client.crt",
        "keyPath": "/certs/client.key"
      }
    },
    "northbound": {
      "endpoint": "xxxx-ats.iot.us-east-1.amazonaws.com",
      "port": 8883,
      "clientId": "my-component",
      "qos": { "publish": 2, "subscribe": 1 },
      "credentials": {
        "caPath": "/certs/AmazonRootCA1.pem",
        "certPath": "/certs/device.pem.crt",
        "keyPath": "/certs/private.pem.key"
      }
    },
    "requestTimeoutSeconds": 30
  }
}

Key rules (all verified, identical across the four languages):

• northbound is optional. Omit it for a local-only (“single-broker”) deployment — the northbound methods are simply unused.
• Local-broker TLS is keyed solely on caPath presence. TLS is enabled when (and only when) the local credentials.caPath is set (useSSL = credentials != null && caPath != null); if caPath is absent, the connection is plaintext.
  • caPath only ⇒ server-only TLS.
  • caPath + certPath + keyPath ⇒ mutual TLS.
  • A client certPath/keyPath without a caPath stays plaintext (the cert is ignored).
  • username/password apply only to a plaintext (no-caPath) local broker. They are set in the else branch under if (!useSSL), so they are mutually exclusive with TLS: once caPath is present the connection goes TLS-only and any username/password is ignored.
• Northbound is a generic MQTT broker. It is plaintext when caPath is absent, server-TLS when caPath is present, and mutual TLS when certPath + keyPath are also present.
• host is an opaque string — a Kubernetes Service DNS name works unchanged.
• local.type is optional and ignored — it is parsed into the config record but never read; there is no defaulting to "mqtt".
• requestTimeoutSeconds (number, default 30, 0 = off) sets the default request deadline.
• local.qos / northbound.qos set MQTT QoS defaults for operations on that broker without an explicit QoS argument. publish and subscribe accept 0 (at most once), 1 (at least once), or 2 (exactly once).

Tip

For local development, bring up the bundled EMQX broker (plaintext 1883, mutual-TLS 8883):

Linux

docker compose -f test-infra/compose.yaml up -d

Windows

docker compose -f test-infra/compose.yaml up -d

Testing: the service-interface seam

Rust and TypeScript expose a substitutable seam so you can inject a fake transport in tests without a broker. Java and Python do not — they have concrete clients only, so you test against the real MessagingClient (point it at a local EMQX broker, or mock at the provider boundary).

Java

No service interface or DI. Test against the concrete MessagingClient — typically against a local broker. Because MessagingClient uses process-global state, reset/close it between tests so state does not leak.

// no substitutable seam — exercise the real client against a local broker
MessagingClient messaging = gg.getMessaging();

Python

No service interface or DI. Test against the concrete static MessagingClient (against a local broker), and reset its process-global state between tests.

# no substitutable seam — exercise the real client against a local broker
messaging = gg.get_messaging()

Rust

The user-facing type is the MessagingService trait, injected as Arc<dyn MessagingService>. In a test, construct DefaultMessagingService::new(provider) with a fake MessagingProvider, or supply your own Arc<dyn MessagingService>.

use std::sync::Arc;
use edgecommons::messaging::{DefaultMessagingService, MessagingService};

let svc: Arc<dyn MessagingService> =
    Arc::new(DefaultMessagingService::new(fake_provider));
svc.publish("test/topic", &msg).await?;

TypeScript

The user-facing type is the IMessagingService interface. In a test, construct new DefaultMessagingService(provider) with a fake MessagingProvider, or pass any object implementing IMessagingService.

import { DefaultMessagingService } from "@edgecommons/edgecommons";

const svc: IMessagingService = new DefaultMessagingService(fakeProvider);
await svc.publish("test/topic", msg);

Caution

Always unsubscribe and handle SIGTERM before your component exits. A run that leaks subscriptions or threads can trip the shared-connection quota on Greengrass.

See also

Unified Namespace guideThe topic grammar, the eight message classes, identity, and the consumer wildcard pattern.

API reference: MessagingThe full public messaging surface — every method, signature, and type, per language.

API reference: UNSgg.uns(), gg.instance(), MessageIdentity, and the validation rules + error codes.

Configuration schema: messaging sectionThe schema for the messaging config section - brokers, requestTimeoutSeconds, and QoS.