Messaging API

This reference covers the principal public messaging APIs in each language. Signature blocks are selected excerpts; they do not reproduce every overload or metadata accessor. For the conceptual walkthrough (transports, wire envelope, concurrency caps and broker config), read the Messaging guide first.

Java is canonical; Python, Rust, and TypeScript mirror its shared contract. The same publish/subscribe, request/reply, and message envelope work identically across all four languages, so components written in any language interoperate on the same topics.

Caution

Several signatures diverge by language in ways that silently change behavior — most notably the subscribe integer arguments (order and meaning), the request/reply future type, and the MessageBuilder method names. Each is called out inline below; the Messaging guide collects them in one place.

Topics are minted through the UNS builder (gg.uns() / gg.instance(id).uns()); the messaging client itself takes literal topics and enforces one rule — the reserved-class publish guard: a client-chosen publish/request/reply to ecv1/.../{state|metric|cfg|log} is rejected (ReservedTopicException / ReservedTopicError / EdgeCommonsError::ReservedTopic).

Accessor

How a component reaches the messaging service from its EdgeCommons handle.

Java

Returns a MessagingClient instance (instance methods).

MessagingClient getMessaging();   // EdgeCommons.getMessaging()

Python

Returns the MessagingClient class — its methods are @staticmethod, called on the class.

def get_messaging() -> MessagingClient: ...   # EdgeCommons.get_messaging()

Rust

Returns Result<Arc<dyn MessagingService>> — errors if no transport was wired.

fn messaging(&self) -> Result<Arc<dyn MessagingService>>;   // EdgeCommons::messaging()

TypeScript

Returns an IMessagingService — throws if no transport was wired.

messaging(): IMessagingService;   // EdgeCommons.messaging()

The messaging client / service

The primary type carrying all publish/subscribe/request/reply methods. Plain methods target the local broker (or Greengrass IPC); northbound methods target the configured northbound transport and take a QoS. The northbound transport is a generic MQTT broker in standalone mode and the Greengrass IoT Core bridge in Greengrass mode.

Java

MessagingClient — instance methods. Built internally via MessagingClientBuilder.create(parsedCommandLine).withReceiveOwnMessages(boolean).build() (receiveOwnMessages defaults to true); you normally obtain it from gg.getMessaging(). receiveOwnMessages is consumed only by the Greengrass IPC provider (it selects the IPC ReceiveMode — RECEIVE_ALL_MESSAGES vs RECEIVE_MESSAGES_FROM_OTHERS); the standalone/MQTT provider takes no such flag and ignores it (an MQTT broker has no own-message filter).

public static final int DEFAULT_MAX_MESSAGES = 10_000;

void publish(String topic, Message msg);
void publishNorthbound(String topic, Message msg, Qos qos);
void publishRaw(String topic, JsonObject metricObject);
void publishNorthboundRaw(String topic, JsonObject metricObject, Qos qos);

void subscribe(String topicFilter, BiConsumer<String,Message> callback);
void subscribe(String topicFilter, BiConsumer<String,Message> callback, int maxConcurrency);
void subscribe(String topicFilter, BiConsumer<String,Message> callback, int maxConcurrency, int maxMessages);
void subscribeNorthbound(String topicFilter, BiConsumer<String,Message> callback, Qos qos);
void subscribeNorthbound(String topicFilter, BiConsumer<String,Message> callback, Qos qos, int maxConcurrency);
void subscribeNorthbound(String topicFilter, BiConsumer<String,Message> callback, Qos qos, int maxConcurrency, int maxMessages);

ReplyFuture request(String topic, Message request);                     // default deadline (30 s)
ReplyFuture request(String topic, Message request, Duration timeout);   // null = default, ZERO = off
ReplyFuture requestNorthbound(String topic, Message request);
ReplyFuture requestNorthbound(String topic, Message request, Duration timeout);
void cancelRequest(ReplyFuture replyFuture);
void cancelRequestNorthbound(ReplyFuture replyFuture);
void reply(Message request, Message reply);
void replyNorthbound(Message request, Message reply);

void unsubscribe(String topicFilter);
void unsubscribeNorthbound(String topicFilter);

static boolean topicMatchesFilter(String topicFilter, String topic);
boolean connected();
void close();
Object getNativeLocalClient();
Object getNativeNorthboundClient();

ReservedPublisher reservedPublisher();   // library-internal: the privileged publish seam
                                         // the guard does not apply to (heartbeat/metric/cfg)

The 3-arg subscribe overload’s int is maxConcurrency, NOT maxMessages (the 4-arg overload adds maxMessages last). ReservedPublisher is public-reachable but documented library-internal — component code uses the normal publish methods.

Python

MessagingClient — every method is a @staticmethod (call on the class returned by get_messaging()). receive_own_messages defaults to False here (Java’s default is True).

def init(args, standalone_config_path=None, receive_own_messages=False) -> MessagingProvider: ...
def shutdown() -> None: ...          # idempotent
def get_messaging_provider(): ...
def connected() -> bool: ...         # never raises

def publish(topic: str, msg: Message) -> None: ...
def publish_raw(topic: str, msg: dict) -> None: ...
def publish_northbound(topic: str, msg: Message, qos: Qos) -> None: ...
def publish_northbound_raw(topic: str, msg: dict, qos: Qos) -> None: ...

def subscribe(topic, callback, max_concurrency=None, max_messages=None) -> None: ...
def subscribe_northbound(topic, callback, qos: Qos, max_concurrency=None, max_messages=None) -> None: ...

def unsubscribe(topic) -> None: ...
def unsubscribe_northbound(topic) -> None: ...

# timeout_secs: None = the configured default (30 s), 0 = disabled for this call
def request(topic, msg, timeout_secs: float | None = None) -> Iou: ...
def request_northbound(topic, msg, timeout_secs: float | None = None) -> Iou: ...
def cancel_request(iou) -> Iou: ...
def cancel_request_northbound(iou) -> Iou: ...
def reply(request, reply) -> None: ...
def reply_northbound(request, reply) -> None: ...

def topic_matches_sub(sub, topic) -> bool: ...
def get_native_client(): ...         # -> {'local': ..., 'northbound': ...}

The 3rd positional arg of subscribe is max_concurrency (matching Java); max_messages is 4th. There is a single get_native_client() (returning a {'local', 'northbound'} dict), not separate local/northbound getters. The library-internal privileged publish seam is the _publish_reserved* staticmethods (underscore convention — not for component code).

Rust

The user-facing seam is the MessagingService trait, obtained as Arc<dyn MessagingService>. It is an async_trait; everything is async except cancel_request* and connected.

#[async_trait]
pub trait MessagingService: Send + Sync {
    async fn publish(&self, topic: &str, msg: &Message) -> Result<()>;
    async fn publish_northbound(&self, topic: &str, msg: &Message, qos: Qos) -> Result<()>;
    async fn publish_raw(&self, topic: &str, payload: &Value) -> Result<()>;
    async fn publish_northbound_raw(&self, topic: &str, payload: &Value, qos: Qos) -> Result<()>;

    // NOTE: max_messages THEN max_concurrency (opposite order from Java/Python)
    async fn subscribe(&self, filter: &str, handler: Arc<dyn MessageHandler>,
                       max_messages: usize, max_concurrency: usize) -> Result<()>;
    async fn subscribe_northbound(&self, filter: &str, handler: Arc<dyn MessageHandler>,
                       qos: Qos, max_messages: usize, max_concurrency: usize) -> Result<()>;

    async fn unsubscribe(&self, filter: &str) -> Result<()>;
    async fn unsubscribe_northbound(&self, filter: &str) -> Result<()>;

    // request() applies the configured default deadline (30 s);
    // request_with_timeout: None = default, Some(Duration::ZERO) = disabled
    async fn request(&self, topic: &str, msg: Message) -> Result<ReplyFuture>;
    async fn request_with_timeout(&self, topic: &str, msg: Message,
                                  timeout: Option<Duration>) -> Result<ReplyFuture>;
    async fn request_northbound(&self, topic: &str, msg: Message) -> Result<ReplyFuture>;
    async fn request_northbound_with_timeout(&self, topic: &str, msg: Message,
                                  timeout: Option<Duration>) -> Result<ReplyFuture>;
    async fn reply(&self, request: &Message, reply: Message) -> Result<()>;
    async fn reply_northbound(&self, request: &Message, reply: Message) -> Result<()>;

    fn cancel_request(&self, reply_future: ReplyFuture);
    fn cancel_request_northbound(&self, reply_future: ReplyFuture);

    fn connected(&self) -> bool;
}

The library-internal privileged publish seam is the crate-private ReservedMessaging trait — the only language where the reserved-class guard is compiler-enforced.

subscribe takes max_messages first, then max_concurrency — the opposite order from Java/Python. The default implementation is DefaultMessagingService::new(provider: Arc<dyn MessagingProvider>) -> Self.

A subscription handler must be an Arc<dyn MessageHandler>. Wrap a closure with the message_handler helper — a bare closure is not accepted:

pub trait MessageHandler: Send + Sync {
    async fn handle(&self, topic: String, message: Message);
}

pub fn message_handler<F, Fut>(f: F) -> Arc<dyn MessageHandler>
where /* F: Fn(String, Message) -> Fut, Fut: Future<Output = ()> */;

// usage:
message_handler(|topic, msg| async move { /* ... */ })

A free function topic_matches(filter, topic) -> bool is exported from the module.

TypeScript

The user-facing seam is the IMessagingService interface, implemented by DefaultMessagingService. Construct it with new DefaultMessagingService(provider: MessagingProvider).

interface IMessagingService {
  publish(topic: string, msg: Message): Promise<void>;
  publishNorthbound(topic: string, msg: Message, qos?: Qos): Promise<void>;
  publishRaw(topic: string, payload: unknown): Promise<void>;
  publishNorthboundRaw(topic: string, payload: unknown, qos?: Qos): Promise<void>;

  // NOTE: maxMessages THEN maxConcurrency; defaults 32 and 1
  subscribe(filter: string, handler: MessageHandler,
            maxMessages?: number, maxConcurrency?: number): Promise<void>;
  subscribeNorthbound(filter: string, handler: MessageHandler,
            qos?: Qos, maxMessages?: number, maxConcurrency?: number): Promise<void>;

  unsubscribe(filter: string): Promise<void>;
  unsubscribeNorthbound(filter: string): Promise<void>;

  // timeoutMs: omitted = the configured default (30 s), explicit 0 = disabled
  request(topic: string, msg: Message, timeoutMs?: number): ReplyFuture;   // synchronous return
  requestNorthbound(topic: string, msg: Message, timeoutMs?: number): ReplyFuture;
  reply(request: Message, reply: Message): Promise<void>;
  replyNorthbound(request: Message, reply: Message): Promise<void>;

  cancelRequest(reply: ReplyFuture): void;
  cancelRequestNorthbound(reply: ReplyFuture): void;

  connected(): boolean;
}

type MessageHandler = (topic: string, message: Message) => void | Promise<void>;

subscribe takes maxMessages first, then maxConcurrency (defaults 32 and 1). request returns the ReplyFuture synchronously; an omitted timeoutMs resolves to the configured messaging.requestTimeoutSeconds default (30 s), and an explicit 0 disables the deadline. subscribeNorthbound’s qos defaults to Qos.AtLeastOnce; the Java/Python northbound methods require an explicit QoS. The library-internal privileged publish seam is the @internal reserved publish on DefaultMessagingService (stripped from the published typings).

Danger

subscribe integer arguments differ in order AND meaning. Java/Python: subscribe(topic, cb, maxConcurrency[, maxMessages]) — the first int is maxConcurrency. Rust/TypeScript: subscribe(filter, handler, maxMessages, maxConcurrency) — the first int is maxMessages. A trailing , 3 means 3 concurrent handlers in Java/Python but queue depth 3 in TypeScript.

Concurrency and queue semantics

• maxConcurrency — maximum concurrent handler invocations for the subscription. Java/Python treat <= 0 as uncapped; Rust/TypeScript clamp that value to 1.
• maxMessages — bounded in-memory queue depth per subscription; on overflow, new messages are dropped with a warning (the subscription never blocks). Default 10000 in Java/Python (DEFAULT_MAX_MESSAGES), 32 in TypeScript, caller-supplied in Rust.
• connected() reports the local broker link only (a northbound outage does not flip it); it backs /readyz, never /livez.

The Message envelope

Normal MQTT and Greengrass IPC messages carry the shared protobuf EdgeCommonsMessage envelope. Its JSON projection has header, optional identity and tags, and body, with optional content type, content encoding and schema metadata. Projection header keys are snake_case; timestamp_ms is the epoch-millisecond timestamp carried by protobuf. Config-bound builders stamp the publisher identity. Component scope omits identity.instance.

Normal subscriptions use the byte decoders below. Foreign JSON, arbitrary bytes and malformed protobuf are rejected, not converted to a synthetic raw message. JSON object helpers remain useful for application tooling and explicit raw publication, but are separate from transport decoding.

Bodies can be typed standard messages, dynamic structured values, or opaque bytes. The bounded opaque-body APIs support 64 KiB. In JSON-facing body views, native bytes may appear as an _edgecommonsBinary base64 marker; protobuf carries the bytes directly. Binary accessors recover the original bytes. Diagnostic JSON may replace opaque content with a length/hash summary, so it is not a general lossless import/export format.

The following are selected API excerpts. Message metadata also includes content type, content encoding, schema and body case; use the language’s accessors when those distinctions matter.

Java

Message exposes accessors; use MessageBuilder for construction.

byte[] toBytes();                  // protobuf transport encoder
static Message fromBytes(byte[] bytes); // protobuf transport decoder
JsonObject toDict();               // JSON-facing projection
JsonObject toDiagnosticJson();     // diagnostic projection (opaque content is summarized)
String getCorrelationId();         // lazily generates a UUID if null
MessageHeader getHeader();
MessageIdentity getIdentity();     // null when the message carries no identity
MessageTags getTags();
void injectTag(String key, String value);
Object getBody();
boolean isBinaryBody();
byte[] getBinaryBody();            // null when not binary; validates the 64 KiB limit
byte[] getOpaqueBody();
String getContentType();
String getContentEncoding();
MessageBodySchema getSchema();
MessageBodyCase getBodyCase();
Object getRaw();
String makeRequest();              // also: makeRequest(String replyTo)
void setCorrelationId(String id);

// Deprecated static factories — prefer MessageBuilder:
static Message buildFromConfig(...);   // @Deprecated
static Message build(Object contents);  // @Deprecated

A non-JsonElement payload (Map/POJO/List) passed to the builder is converted via the Gson reflective tree, so withPayload(Map) works.

Python

Message is a dataclass with optional header, identity, tags and content metadata, plus its body.

def to_bytes(self) -> bytes: ...       # protobuf transport encoder
@staticmethod
def from_bytes(payload: bytes) -> "Message": ...
def to_diagnostic_json(self) -> dict: ...
def to_dict(self) -> dict: ...         # JSON-facing projection
def __str__(self) -> str: ...
def dumps(self, indent=None) -> str: ...
def get_correlation_id(self): ...
def get_header(self) -> MessageHeader: ...
def get_identity(self) -> Optional[MessageIdentity]: ...
def get_tags(self) -> MessageTags: ...
def get_body(self): ...
def is_binary_body(self) -> bool: ...
def get_binary_body(self) -> Optional[bytes]: ...   # None when body is not binary; validates marker + 64 KiB limit
def get_raw(self): ...
def get_source(self): ...           # alias of get_tags
def get_payload(self): ...          # alias of get_body
def inject_tag(self, key, value): ...
def make_request(self): ...
def set_correlation_id(self, id): ...

@staticmethod
def from_object(msg_contents) -> "Message": ...

Rust

pub struct Message {
    // Selected fields; additional content metadata is omitted from this excerpt.
    pub header: MessageHeader,
    pub identity: Option<MessageIdentity>,   // skipped when None
    pub tags: Option<MessageTags>,           // skipped when None
    pub body: Value,
    pub raw: Option<Value>,
}

impl Message {
    pub fn raw(value: Value) -> Self;
    pub fn is_raw(&self) -> bool;
    pub fn get_raw(&self) -> Option<&Value>;
    pub fn is_binary_body(&self) -> bool;
    pub fn binary_body(&self) -> Result<Option<Vec<u8>>>;   // validates marker + 64 KiB limit
    pub fn to_vec(&self) -> Result<Vec<u8>>;              // protobuf encoder
    pub fn from_slice(bytes: &[u8]) -> Result<Message>;   // strict protobuf decoder
    pub fn body_case(&self) -> MessageBodyCase;
    pub fn opaque_body(&self) -> Result<Option<Vec<u8>>>;
    pub fn correlation_id(&self) -> &str;
}

impl MessageBuilder {
    pub fn binary_payload(self, bytes: impl AsRef<[u8]>) -> Result<Self>;
}

Custom Serde Serialize/Deserialize handles the JSON-facing representation, including explicit raw values and optional metadata. It is separate from to_vec/from_slice, which encode/decode protobuf. Raw values are not valid input to the protobuf encoder.

TypeScript

class Message {
  header: MessageHeader;
  identity?: MessageIdentity;   // undefined when the message carries no identity
  tags?: MessageTags;
  body: unknown;
  contentType?: string;
  contentEncoding?: string;
  schema?: MessageBodySchema;
  bodyCase?: MessageBodyCase;

  static raw(value: unknown): Message;
  static fromObject(value: unknown): Message;
  static fromWire(data: Buffer | string): Message; // JSON helper; non-JSON becomes raw text
  static fromBytes(data: Buffer | Uint8Array): Message; // strict protobuf decoder

  isRaw(): boolean;
  getRaw(): unknown;
  getBody(): unknown;
  isBinaryBody(): boolean;
  getBinaryBody(): Buffer | undefined;   // validates marker + 64 KiB limit
  getCorrelationId(): string;
  getReplyTo(): string | undefined;
  toObject(): Record<string, unknown>;  // JSON-facing projection
  toJSON(): string;                     // JSON text, not transport bytes
  toBytes(): Buffer;                    // protobuf encoder
  toDiagnosticJson(): Record<string, unknown>;
}

MessageHeader and MessageTags

The header carries message metadata. timestamp_ms is the canonical protobuf epoch-millisecond value; timestamp is its RFC3339 projection/API representation. tags carries business context only (your custom tags config keys). tags has no special thing handling in any language — the publisher’s device travels in the top-level identity element, and a stray inbound thing key just lands in the generic tag map. The reply-topic prefix is the literal string edgecommons/reply- (trailing -, not /) in all four languages; reply topics are deliberately non-UNS.

Java

class MessageHeader {
    String name; String version; String timestamp;   // ISO-8601 / RFC3339 (ISO_INSTANT)
    long timestampMs;
    String correlationId; String uuid; String replyTo;
    static final String REPLY_MESSAGE_TOPIC_PREFIX = "edgecommons/reply-";
    JsonObject toDict();   // projection includes timestamp_ms, name, version, timestamp, uuid, correlation_id, [reply_to]
}

class MessageTags {
    JsonObject tags;                           // the free-form tag map (no thing field)
    JsonObject toDict();                       // flattens the tag map
    static MessageTags fromConfig(ConfigManager config);
    void injectTag(String key, String value);
}

Python

@dataclass
class MessageHeader:
    name: str
    version: str
    correlation_id: Optional[str] = None
    timestamp: Optional[str] = None   # ISO-8601 / RFC3339 string
    timestamp_ms: Optional[int] = None
    uuid: Optional[str] = None
    reply_to: Optional[str] = None
    REPLY_MESSAGE_TOPIC_PREFIX = "edgecommons/reply-"
    # __post_init__ fills timestamp/correlation_id/uuid
    def to_dict(self) -> dict: ...   # includes timestamp_ms and snake_case metadata

@dataclass
class MessageTags:
    tags: dict = field(default_factory=dict)   # free-form tag map (no thing field)
    def to_dict(self) -> dict: ...

Rust

pub struct MessageHeader {
    pub name: String,
    pub version: String,
    pub timestamp: String,   // RFC3339 string
    pub timestamp_ms: u64,
    pub correlation_id: String,
    pub uuid: String,
    pub reply_to: Option<String>,   // serde skips when None
}

pub struct MessageTags {
    pub extra: BTreeMap<String, Value>,   // serde flatten — the free-form tag map
}

TypeScript

interface MessageHeader {
  name: string;
  version: string;
  timestamp: string;   // ISO-8601 string (Date.toISOString())
  timestamp_ms?: number;
  correlation_id: string;
  uuid: string;
  reply_to?: string;
}

type MessageTags = Record<string, unknown>;   // free-form tag map (no thing key)

For the MessageIdentity type (hier/path/component/instance, the computed device accessor, and withInstance), see the UNS API reference.

MessageBuilder

Construct messages with the builder, not raw constructors — it stamps uuid, correlation_id, and timestamp for you. Supplying the config is what stamps the custom tags and the UNS identity (with the optional per-message instance token — omit it for a component-scoped message); build() never requires config in any language (a config-less build produces a legal envelope without identity). An explicit identity override always wins over the config-resolved one. All four builders expose explicit uuid/timestamp setters (deterministic envelopes for tests and vectors).

Caution

Method names diverge. Java/Python/TypeScript use create / withPayload / withConfig (Python: with_payload / with_config); Rust uses new / payload / from_config and drops the with_ prefix throughout.

Java

static MessageBuilder create(String name, String version);
static Message fromObject(Object msgContents);     // classifies dict→envelope vs raw

MessageBuilder withCorrelationId(String id);
MessageBuilder withUuid(String uuid);
MessageBuilder withTimestamp(String timestamp);
MessageBuilder withPayload(Object payload);         // JsonObject / Map / POJO / List
MessageBuilder withConfig(ConfigManager config);    // stamps tags + the UNS identity
MessageBuilder withInstance(String instance);       // per-message instance token (omit for component scope)
MessageBuilder withIdentity(MessageIdentity id);    // explicit override (tests, vectors, relays)
Message build();

A string payload that parses as JSON is stored as the parsed object; otherwise it is stored as a raw string. The companion MessageHeaderBuilder.create(name, version) offers .withCorrelationId/.withTimestamp/.withUuid/.withReplyTo and .build().

Python

@staticmethod
def create(name, version) -> MessageBuilder: ...
@staticmethod
def from_object(msg_contents) -> Message: ...

def with_correlation_id(self, id): ...
def with_payload(self, body): ...
def with_config(self, config): ...      # stamps tags + the UNS identity
def with_tags(self, tags): ...          # explicit tags (alternative to config)
def with_uuid(self, uuid): ...
def with_timestamp(self, ts): ...
def with_reply_to(self, topic): ...
def with_instance(self, instance): ...  # per-message instance token (omit for component scope)
def with_identity(self, identity): ...  # explicit override (tests, vectors, relays)
def build(self) -> Message: ...

Rust

Different method names — new (not create), payload (not with_payload), from_config (not with_config):

impl MessageBuilder {
    pub fn new(name: impl Into<String>, version: impl Into<String>) -> Self;
    pub fn payload(self, body: Value) -> Self;
    pub fn correlation_id(self, id: impl Into<String>) -> Self;
    pub fn uuid(self, uuid: impl Into<String>) -> Self;
    pub fn timestamp(self, timestamp: impl Into<String>) -> Self;
    pub fn reply_to(self, topic: impl Into<String>) -> Self;
    pub fn tag(self, key: impl Into<String>, value: Value) -> Self;
    pub fn from_config(self, config: &Config) -> Self;      // stamps tags + the UNS identity
    pub fn instance(self, instance: impl Into<String>) -> Self;   // per-message token (omit for component scope)
    pub fn identity(self, identity: MessageIdentity) -> Self;     // explicit override
    pub fn build(self) -> Message;
}

There is no thing_name(...) setter; the publisher’s device travels in the UNS identity.

TypeScript

class MessageBuilder {
  static create(name: string, version: string): MessageBuilder;
  withPayload(body: unknown): this;
  withTags(tags: MessageTags): this;
  withTag(key: string, value: unknown): this;
  withConfig(config: {
    parsed: { tags: Record<string, unknown> };
    componentIdentity?: MessageIdentity;
  }): this;                                        // stamps tags + the UNS identity
  withCorrelationId(id: string): this;
  withUuid(uuid: string): this;
  withTimestamp(timestamp: string): this;
  withReplyTo(topic: string): this;
  withInstance(instance: string): this;            // per-message instance token (omit for component scope)
  withIdentity(identity: MessageIdentity): this;   // explicit override
  build(): Message;
}

There is no withThingName(...) setter; the publisher’s device travels in the UNS identity. withConfig is typed structurally so it accepts the Config snapshot without a config import.

Request/reply future and the internal deadline

request returns a future that resolves when the matching reply arrives. The responder’s reply(request, reply) copies the request’s correlation_id onto the reply before publishing it to the request’s reply_to topic.

Every request arms a framework-owned deadline at send time — the configured messaging.requestTimeoutSeconds default (30 s; 0 disables) unless a per-call value is given. When it fires, the library (1) unsubscribes the ephemeral reply topic, (2) removes the pending entry, and (3) completes the future exceptionally — even if the caller never awaits it, so an abandoned request cannot leak its reply subscription. Reply-arrival, the deadline, and cancelRequest race through one idempotent settle path; a straggler reply after settle is logged at DEBUG and dropped. As a result, a no-arg get()/await is bounded by the deadline, and a get(t) waits at most min(t, deadline).

Language	Per-call override	Deadline failure signal
Java	request(topic, msg, Duration) (null = default, ZERO = off)	java.util.concurrent.TimeoutException (exceptional completion)
Python	request(topic, msg, timeout_secs=...) (None = default, 0 = off)	Iou.get() raises RequestTimeoutError
Rust	request_with_timeout(topic, msg, Option<Duration>)	future resolves Err(EdgeCommonsError::RequestTimeout { .. })
TypeScript	request(topic, msg, timeoutMs) (omitted = default, 0 = off)	promise rejects with RequestTimeoutError

Java

ReplyFuture extends CompletableFuture<Message> — use the standard future API (get(), orTimeout(), thenAccept()). It exposes a public String replyTopic; field.

public class ReplyFuture extends CompletableFuture<Message> {
    public String replyTopic;
}

// usage:
ReplyFuture future = messaging.request("svc/op", request);   // 30 s default deadline
Message reply = future.get();                                 // bounded by the deadline
// on deadline: ExecutionException wrapping java.util.concurrent.TimeoutException

Python

request returns an Iou. Its get(timeout) returns a (done, result) tuple — not the reply directly. timeout < 0 blocks (up to the framework deadline); on a local wait timeout it returns (False, self). When the framework deadline settled the request, get() raises RequestTimeoutError.

class Iou:
    def get(self, timeout=-1) -> Tuple[bool, Any]: ...   # (done, result); raises on deadline
    def set_result(self, result) -> None: ...
    def set_error(self, error) -> None: ...              # exceptional completion
    def done(self) -> bool: ...
    def get_user_data(self): ...

# usage:
iou = messaging.request("svc/op", request)               # 30 s default deadline
done, reply = iou.get(timeout=5)                         # may raise RequestTimeoutError

Rust

request() returns Result<ReplyFuture>; await the call to get the future, then await the future. A supervisor task owns the reply subscription and races the reply against the deadline; dropping the ReplyFuture still cancels the request and unsubscribes the reply topic.

// ReplyFuture: impl Future<Output = Result<Message>>
let fut = messaging.request("svc/op", request).await?;    // 30 s default deadline
let reply = fut.await?;                                    // Err(EdgeCommonsError::RequestTimeout) on deadline

TypeScript

ReplyFuture implements PromiseLike<Message> — await it directly. The timeout is the third argument to request (omitted = the configured default; explicit 0 disables). It also exposes .cancel().

interface ReplyFuture extends PromiseLike<Message> {
  cancel(): void;
}

// usage:
const reply = await messaging.request("svc/op", req);     // 30 s default deadline
// on deadline: rejects with RequestTimeoutError

Qos and Destination enums

MQTT QoS defaults are configured under each broker, messaging.local.qos and messaging.northbound.qos, for operations that do not carry an explicit QoS parameter. The standalone local and northbound MQTT brokers support QoS 0/1/2: 0 is at most once, 1 is at least once, and 2 is exactly once. All four language APIs expose EdgeCommons-owned QoS enums. Greengrass IoT Core IPC accepts only QoS 0/1; calls using QoS 2 fail at the IPC provider boundary. Direct northbound methods that take a QoS argument still use that per-call value.

Destination exists in Rust/TS as a Local / Northbound selector used by the lower-level provider (Rust Destination::Northbound, TypeScript Destination.Northbound — string value "northbound" in both); Java/Python do not expose an equivalent enum because they use per-destination method pairs.

Java

Qos is the Java-native EdgeCommons enum. There is no separate Destination enum — use the plain vs *Northbound method pairs.

import com.mbreissi.edgecommons.messaging.Qos;

Qos.AT_MOST_ONCE;    // MQTT QoS 0
Qos.AT_LEAST_ONCE;   // MQTT QoS 1
Qos.EXACTLY_ONCE;    // MQTT QoS 2; standalone MQTT only

Python

Qos is the Python-native EdgeCommons enum. No separate Destination enum — use the plain vs *_northbound method pairs.

from edgecommons.messaging.qos import Qos

Qos.AT_MOST_ONCE     # MQTT QoS 0
Qos.AT_LEAST_ONCE    # MQTT QoS 1
Qos.EXACTLY_ONCE     # MQTT QoS 2; standalone MQTT only

Rust

Qos is a Rust-native enum. Destination is used only by the lower-level MessagingProvider trait.

pub enum Qos { AtMostOnce, AtLeastOnce, ExactlyOnce }
pub enum Destination { Local, Northbound }   // RFC-430 acronym casing (Rust idiom)

TypeScript

Qos and Destination are TS-native string enums.

enum Qos { AtMostOnce = "atMostOnce", AtLeastOnce = "atLeastOnce", ExactlyOnce = "exactlyOnce" }
enum Destination { Local = "local", Northbound = "northbound" }

The provider / interface seam

Below the client/service sits a transport MessagingProvider. Rust and TypeScript expose a substitutable seam — the MessagingService trait / IMessagingService interface plus a swappable provider — so you can inject a fake transport in tests. Java and Python have concrete clients only (no service interface, no DI); the two providers are internal.

Java

No service interface or DI. The abstract base MessagingProvider has two concrete subclasses — GreengrassMessagingProvider (IPC) and StandaloneMessagingProvider (dual-MQTT). There is no single-client MqttProvider class. Test against the concrete MessagingClient (point it at a local broker; reset its process-global state between tests).

Python

No service interface or DI. The MessagingProvider ABC (DEFAULT_MAX_MESSAGES = 10000) has two concrete subclasses — GreengrassIpcProvider and StandaloneProvider. There is no MqttProvider. Test against the concrete static MessagingClient.

Rust

MessagingService (above) is the user-facing trait. Below it, the lower-level MessagingProvider trait carries raw transport:

#[async_trait]
pub trait MessagingProvider: Send + Sync {
    async fn publish(&self, topic: &str, payload: Vec<u8>, dest: Destination, qos: Qos) -> Result<()>;
    async fn subscribe(&self, filter: &str, dest: Destination, qos: Qos, max_messages: usize) -> Result<Subscription>;
    async fn unsubscribe(&self, filter: &str, dest: Destination) -> Result<()>;
    fn connected(&self) -> bool;
}

// Inject a fake provider in tests:
let svc: Arc<dyn MessagingService> = Arc::new(DefaultMessagingService::new(fake_provider));

subscribe returns a Subscription — a bounded, polled queue of (topic, payload) messages (max_messages cap), not a handler callback.

TypeScript

IMessagingService (above) is the user-facing interface. Below it, the MessagingProvider interface carries raw bytes:

interface MessagingProvider {
  publishBytes(topic: string, payload: Buffer, dest: Destination, qos: Qos): Promise<void>;
  subscribeRaw(
    filter: string,
    dest: Destination,
    qos: Qos,
    onMessage: (topic: string, payload: Buffer) => void,
  ): Promise<RawSubscription>;
  connected(): boolean;
  disconnect(): Promise<void>;
}

// Inject a fake provider in tests:
const svc: IMessagingService = new DefaultMessagingService(fakeProvider);

Messaging (broker) config types

With the MQTT transport, the broker config is a JSON messaging section with a local block and an optional northbound block, plus the request-deadline default. QoS defaults belong inside each broker block. The wire shape is identical across the four languages (each maps it to its own config type — Java MessagingConfiguration, Python messaging_config.py dataclasses, Rust config.rs, TS config.ts).

{
  "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",
        "username": "optional-user",
        "password": "optional-pass"
      }
    },
    "northbound": {
      "endpoint": "northbound-broker.example.com",
      "port": 8883,
      "clientId": "my-component",
      "qos": { "publish": 1, "subscribe": 1 },
      "credentials": {
        "caPath": "/certs/northbound-ca.pem",
        "certPath": "/certs/device.pem.crt",
        "keyPath": "/certs/private.pem.key"
      }
    },
    "requestTimeoutSeconds": 30
  }
}

Field reference (verified, identical across all four languages):

• local — host (string), port (int), clientId (string), optional qos and credentials.
• northbound — host or endpoint (string), port (int), clientId (string), optional qos and credentials. The whole northbound block is optional: omit it for a local-only (“single-broker”) deployment.
• credentials — caPath, certPath, keyPath, and optional username / password.
• requestTimeoutSeconds — number, min 0, default 30; the default request deadline. 0 disables it; an explicit per-call timeout always wins.
• local.qos / northbound.qos — optional 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).

Rules:

• Local-broker TLS is keyed solely on caPath presence (useSSL = credentials != null && caPath != null). caPath only ⇒ server-only TLS; caPath + certPath + keyPath ⇒ mutual TLS; a certPath/keyPath without 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 and ignored once caPath is present.
• A TLS northbound broker may require full mutual-TLS credentials (caPath + certPath + keyPath).
• host is an opaque string — a Kubernetes Service DNS name works unchanged.
• local.type is optional and ignored — parsed into the config record but never read; there is no defaulting to "mqtt".
• The broker file is parsed leniently (plain Gson in Java) and is not schema-validated at runtime; unknown/mistyped keys are silently ignored.

End-to-end example

A minimal publish, subscribe, and request/reply round trip in each language.

Java

MessagingClient messaging = gg.getMessaging();

// subscribe — 3rd arg is maxConcurrency
messaging.subscribe("itest/pubsub", (topic, m) -> {
    System.out.println(m.getHeader().getName() + " -> " + m.getBody());
}, 1);

// publish
Message msg = MessageBuilder.create("Hello", "1.0")
    .withPayload(payload)        // JsonObject / Map / POJO
    .withConfig(configManager)   // stamps tags + the UNS identity
    .build();
messaging.publish("itest/pubsub", msg);

// request/reply — ReplyFuture extends CompletableFuture<Message>;
// a framework deadline (default 30 s) bounds the wait
ReplyFuture future = messaging.request("itest/request", request);
Message reply = future.get();
// responder side:
messaging.reply(request, replyMsg);   // copies correlation id

Python

messaging = gg.get_messaging()           # the MessagingClient class (static methods)

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

messaging.subscribe("evt/topic", handler, 1)   # 3rd arg = max_concurrency

msg = MessageBuilder.create("Hello", "1.0").with_payload({"n": 1}).with_config(config).build()
messaging.publish("evt/topic", msg)

iou = messaging.request("svc/op", request)     # returns an Iou; 30 s framework deadline
done, reply = iou.get(timeout=5)               # (done, result) TUPLE; raises RequestTimeoutError
# responder side:                               # when the framework deadline fired
messaging.reply(request, reply_msg)            # copies correlation id

Rust

use serde_json::json;
use edgecommons::messaging::message_handler;

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

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

let msg = MessageBuilder::new("Hello", "1.0")   // new(), not create()
    .payload(json!({ "n": 1 }))
    .from_config(&gg.config())                   // stamps tags + the UNS identity
    .build();
messaging.publish("itest/pubsub", &msg).await?;

// request/reply: request() -> Result<ReplyFuture>, then await the future
// (a framework deadline — default 30 s — bounds the wait)
let fut = messaging.request("svc/op", request).await?;
let reply = fut.await?;                          // Err(EdgeCommonsError::RequestTimeout) on deadline
// responder side:
responder.reply(&request, reply_msg).await?;    // copies correlation id

TypeScript

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

await messaging.subscribe("evt/topic", (_t, m) => {
  console.log(m.getBody());
}, 32, 1);                                      // maxMessages, maxConcurrency

const msg = MessageBuilder.create("evt", "1.0.0").withPayload({ n: 1 }).build();
await messaging.publish("evt/topic", msg);

// request/reply: request() returns a ReplyFuture you await;
// 3rd arg = timeoutMs (omitted = the 30 s configured default)
const reply = await messaging.request("rpc/echo", req, 1000);
console.log(reply.getBody());
// responder side:
await messaging.reply(req, replyMsg);          // copies correlation id

See also

Messaging guideThe conceptual walkthrough — transports, the envelope, concurrency caps, and the broker config rules.

UNS APIgg.uns(), gg.instance(), MessageIdentity, the validation rules, and the reserved-class guard.

Configuration schemaThe full config schema, including the messaging section.