Messaging API

This is the API reference for the messaging subsystem: every public type and method signature, per language. For the conceptual walkthrough (transports, the wire envelope, concurrency caps, the broker config rules), read the Messaging guide first — this page is the exhaustive surface.

All signatures are verified against source. Java is canonical; Python, Rust, and TypeScript mirror it. 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.

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.

Returns a MessagingClient instance (instance methods).

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

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

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

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

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

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.

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.

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).

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.

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).

Concurrency and queue semantics

maxConcurrency — maximum concurrent handler invocations for the subscription. A value <= 0 is uncapped (Java/Python skip the semaphore; Rust/TS clamp up 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

The on-wire shape is identical across all four languages: { "header": {...}, "identity": {...}, "tags": {...}, "body": <any> }, or { "raw": <value> } for a raw message. Header keys are snake_case; reply_to is omitted when absent. identity is the publisher’s UNS identity (stamped by a config-bound builder; legally absent on bootstrap/raw-bridged messages); tags carries business context only and is emitted only when tags were stamped. A received payload with none of header/identity/tags/body (or non-JSON bytes) is delivered as a raw message; non-JSON bytes become a raw string. A malformed inbound identity parses leniently to null/absent with a warning — the message still delivers.

Binary bodies are first-class but deliberately bounded: a decoded binary body is limited to 64 KiB and travels inside body as:

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

The normal body accessors return that marker object on inbound messages. Use the binary-specific accessors below to decode and validate it.

Message (package-private fields MessageHeader header; MessageIdentity identity; MessageTags tags; Object body; Object raw;).

JsonObject toDict();               // header, identity, tags, body (identity between header and tags)
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 body is not binary; validates marker + 64 KiB limit
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.

Message is a @dataclass(header=None, identity=None, tags=None, body=None, raw=None).

def to_dict(self) -> dict: ...
def __str__(self) -> str: ...
def dumps(self) -> 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": ...

pub struct Message {
    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>>;
    pub fn from_slice(bytes: &[u8]) -> Result<Message>;   // non-JSON → raw string
    pub fn correlation_id(&self) -> &str;
}

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

Custom Serialize/Deserialize: a raw message serializes as {"raw": ..}, otherwise as {"header", "identity", "tags", "body"} (absent identity/tags skipped).

class Message {
  header: MessageHeader;
  identity?: MessageIdentity;   // undefined when the message carries no identity
  tags: MessageTags;
  body: unknown;

  static envelope(header: MessageHeader, tags: MessageTags, body: unknown): Message;
  static raw(value: unknown): Message;
  static fromObject(value: unknown): Message;
  static fromWire(data: Buffer | string): Message;   // non-JSON → raw string

  isRaw(): boolean;
  getRaw(): unknown;
  getBody(): unknown;
  isBinaryBody(): boolean;
  getBinaryBody(): Buffer | undefined;   // validates marker + 64 KiB limit
  getCorrelationId(): string;
  getReplyTo(): string | undefined;
  toObject(): unknown;
  toJSON(): string;
}

`MessageHeader` and `MessageTags`

The header carries message metadata; 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.

class MessageHeader {
    String name; String version; String timestamp;   // ISO-8601 / RFC3339 (ISO_INSTANT)
    String correlationId; String uuid; String replyTo;
    static final String REPLY_MESSAGE_TOPIC_PREFIX = "edgecommons/reply-";
    JsonObject toDict();   // emits 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);
}

@dataclass
class MessageHeader:
    name: str
    version: str
    correlation_id: Optional[str] = None
    timestamp: Optional[str] = None   # ISO-8601 / RFC3339 string
    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: ...   # snake_case: name, version, timestamp, uuid, correlation_id, [reply_to]

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

pub struct MessageHeader {
    pub name: String,
    pub version: String,
    pub timestamp: String,   // RFC3339 string
    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
}

interface MessageHeader {
  name: string;
  version: string;
  timestamp: string;   // ISO-8601 string (Date.toISOString())
  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).

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().

@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: ...

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.

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`

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

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

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

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.

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

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

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)

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.

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).

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.

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.

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.

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

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

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

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