Streaming API

API reference for the streaming subsystem: durable store-and-forward telemetry egress to Kinesis or Kafka. This page enumerates the public surface per language; for the task-oriented walkthrough see the Streaming guide.

Streaming is opt-in: it is active only when the component config has a streaming section, and in Rust only when the streaming cargo feature is enabled. All four languages bind the same shared Rust core (edgestreamlog): Rust links it directly, while Java (FFM / Panama), Python (PyO3), and TypeScript (napi-rs) reach it through a native binding.

In hierarchical-config deployments, streaming.streams may be inherited from a hierarchy layer. The inherited entries do not create a shared runtime service: each component owns its own stream service, buffer files, drains, metrics bridge, and shutdown lifecycle. Disk buffer paths must resolve to component-unique locations.

Accessor

How you get the service from the EdgeCommons runtime handle. The return shape encodes the opt-in contract.

Language	Accessor	Returns
Java	StreamService getStreams()	StreamService, or null if there is no streaming config
Python	get_streams(self)	StreamService, or None if there is no streaming config
Rust	streams(&self) -> Arc<dyn streaming::StreamService>	always an Arc (never None); stream() errors if not configured. Behind #[cfg(feature = "streaming")]
TypeScript	streams(): StreamService | undefined	StreamService, or undefined if there is no streaming config

StreamService

Opens and owns every stream declared in the streaming section. In Java, Python, and TypeScript it is a class with a static open factory; in Rust it is a trait (StreamService) implemented by DefaultStreamService. Java, Python, and TypeScript instances are closeable; the Rust value is dropped via RAII.

Java

com.mbreissi.edgecommons.streaming.StreamService implements AutoCloseable

static StreamService open(String configJson)        // opens every stream; loads the native lib on first use
StreamHandle      stream(String name)               // throws EdgeStreamException (ERR_UNKNOWN_STREAM) if absent
StreamStats       stats(String name)                // throws EdgeStreamException (ERR_UNKNOWN_STREAM) if absent
String            configJson()                      // the resolved config this service was opened with
static List<String> streamNames(String configJson)  // stream names declared in a config, without opening
static void       registerSink(SinkFunction sink)   // BYO-sink for `callback` streams; call BEFORE open()
static void       unregisterSink()
static boolean    nativeAvailable()                 // capability probe; returns false instead of throwing
synchronized void close()                           // flush all + stop engines; idempotent

Python

edgecommons.streaming.StreamService (context-manager)

@staticmethod
def open(config_json: str) -> "StreamService": ...

@staticmethod
def open_with_callback(config_json: str, callback: SinkCallback) -> "StreamService":
    """Bind `callback` as the sink for every {"type": "callback"} stream."""

def stream(self, name: str) -> StreamHandle: ...        # raises EdgeStreamError (ERR_UNKNOWN_STREAM) if absent
def stats(self, name: str) -> StreamStats: ...          # raises EdgeStreamError (ERR_UNKNOWN_STREAM) if absent

@staticmethod
def stream_names(config_json: str) -> List[str]: ...

def close(self) -> None: ...
def __enter__(self) -> "StreamService": ...             # `with StreamService.open(cfg) as svc:`
def __exit__(self, *exc) -> None: ...

Rust

edgecommons::streaming — feature-gated behind the off-by-default streaming cargo feature.

// The substitutable seam:
pub trait StreamService: Send + Sync {
    fn stream(&self, name: &str) -> Result<StreamHandle>;   // Err(EdgeCommonsError::Streaming) if not configured
    fn stream_names(&self) -> Vec<String>;
    fn stats(&self, name: &str) -> Option<Stats>;           // None (not an error) if not configured
}

// The default implementation:
pub struct DefaultStreamService { /* ... */ }
impl DefaultStreamService {
    pub fn open(config: &Config) -> Result<Self>;
    pub fn open_with(config: &Config, sink_factory: &SinkFactory) -> Result<Self>; // custom / test sink
}
// No close(): the service is dropped via RAII (flushes + stops the export engines on drop).

There is no nativeAvailable() — availability is compile-time (see Native core).

TypeScript

edgecommons.StreamService

static open(configJson: string): StreamService;
stream(name: string): StreamHandle;          // throws EdgeStreamError (ERR_UNKNOWN_STREAM) if absent
stats(name: string): StreamStats;            // throws EdgeStreamError (ERR_UNKNOWN_STREAM) if absent
static streamNames(configJson: string): string[];
close(): void;                               // flush all + stop engines

Unconfigured-stream behavior differs in Rust

Java, Python, and TypeScript throw ERR_UNKNOWN_STREAM from stream() and stats() for an unknown name. In Rust, stream() returns Result::Err and stats() returns Option::None (not an error) when the stream is not configured.

StreamHandle

A handle to one configured stream. Append records, force the buffer to disk with flush(), and read the handle name. Java, Python, and TypeScript handles are closeable; the Rust handle is Clone / Arc-backed with no close().

Java

com.mbreissi.edgecommons.streaming.StreamHandle implements AutoCloseable

String name();
void append(String partitionKey, long timestampMs, byte[] payload);
        // partitionKey <= 65535 bytes (IllegalArgumentException otherwise);
        // throws EdgeStreamException on buffer / IO / sink error
void flush();              // force the buffer durably to disk; does NOT wait for export
synchronized void close(); // idempotent; the handle stays usable for append/flush even after service.close()

Python

edgecommons.streaming.StreamHandle (context-manager)

def append(self, partition_key: str, timestamp_ms: int, payload: bytes) -> None: ...
def flush(self) -> None: ...        # force the buffer durably to disk; does NOT wait for export
def close(self) -> None: ...
def __enter__(self) -> "StreamHandle": ...
def __exit__(self, *exc) -> None: ...

Rust

edgecommons::streaming::StreamHandle — #[derive(Clone)], Arc-backed.

pub fn append(&self, rec: Record) -> Result<()>;          // takes one Record value, not 3 args
pub fn append_batch(&self, recs: &[Record]) -> Result<()>; // Rust-only — no batch append elsewhere
pub fn flush(&self) -> Result<()>;                        // force the buffer durably to disk
pub fn name(&self) -> &str;
// No close(): dropped via RAII.

// Record is re-exported as StreamRecord:
pub struct Record { /* partition_key: String, timestamp_ms: u64, payload: Vec<u8> */ }
impl Record {
    pub fn new(
        partition_key: impl Into<String>,
        timestamp_ms: u64,
        payload: impl Into<Vec<u8>>,
    ) -> Record;
}

TypeScript

edgecommons.StreamHandle

readonly name: string;
append(partitionKey: string, timestampMs: number, payload: Buffer | Uint8Array): void;
flush(): void;   // force the buffer durably to disk; does NOT wait for export
close(): void;

The append signature differs in Rust

Java, Python, and TypeScript take three positional arguments — append(partitionKey, timestampMs, payload). Rust takes a single Record value built with StreamRecord::new(partitionKey, timestampMs, payload), and additionally offers a Rust-only append_batch(&[Record]).

Append example

Get the service from the runtime accessor, fetch a StreamHandle by name, append, then flush.

Java

StreamService streams = gg.getStreams();   // null if there is no `streaming` config
if (streams != null) {
    StreamHandle h = streams.stream("telemetry");
    h.append("pump-7", System.currentTimeMillis(),
             "reading-0".getBytes(StandardCharsets.UTF_8));
    h.flush();

    StreamStats s = streams.stats("telemetry");
    log.info("appended={} backlog={}", s.appendedTotal(), s.backlog());
}

Python

streams = gg.get_streams()   # None if there is no `streaming` config
if streams is not None:
    h = streams.stream("telemetry")
    h.append("pump-7", int(time.time() * 1000), b"reading-0")
    h.flush()

    s = streams.stats("telemetry")
    log.info("appended=%s backlog=%s", s.appended_total, s.backlog)

Rust

#[cfg(feature = "streaming")]
{
    use edgecommons::streaming::StreamRecord;

    let streams = gg.streams();             // Arc<dyn StreamService> — always present
    let h = streams.stream("telemetry")?;   // Err(EdgeCommonsError::Streaming) if not configured
    h.append(StreamRecord::new("pump-7", now_ms, b"reading-0".to_vec()))?;
    h.flush()?;

    if let Some(s) = streams.stats("telemetry") {   // None (not an error) if not configured
        tracing::info!(appended = s.appended_total, backlog = s.backlog);
    }
}

TypeScript

const streams = gg.streams();   // StreamService | undefined when there is no `streaming` config
if (streams) {
  const h = streams.stream("telemetry");
  h.append("pump-7", Date.now(), Buffer.from("reading-0"));
  h.flush();

  const s = streams.stats("telemetry");
  console.log(`appended=${s.appendedTotal} backlog=${s.backlog}`);
}

StreamStats

A snapshot of per-stream counters returned by stats(name). The same ten numeric fields exist in all four languages — camelCase in Java and TypeScript, snake_case in Python and Rust. Rust carries one additional field.

Field (camelCase / snake_case)	Type	Meaning
appendedTotal / appended_total	long / int / u64 / number	Records accepted into the buffer.
exportedTotal / exported_total	“	Records sent to the sink.
droppedTotal / dropped_total	“	Records dropped (e.g. by onFull: dropOldest).
retriesTotal / retries_total	“	Export retry attempts.
failedTotal / failed_total	“	Records that failed permanently.
backlog	“	Records buffered but not yet exported.
diskBytes / disk_bytes	“	Bytes currently on disk.
ackedOffset / acked_offset	“	Highest acknowledged log offset.
nextOffset / next_offset	“	Next log offset to be assigned.
oldestUnackedAgeMs / oldest_unacked_age_ms	“	Age of the oldest un-acked record, in ms.

• Java: StreamStats is a record with accessor methods (appendedTotal(), backlog(), …).
• Python: StreamStats is a frozen @dataclass with snake_case attributes.
• Rust: Stats exposes the ten u64 counters plus an 11th field last_export_error: Option<String> (currently always None in the wrapper). The prelude re-exports it as StreamStats.
• TypeScript: StreamStats is an interface with the ten camelCase number fields.

Sink config types

A stream’s sink is tagged by type. Three sink types are declarable in component config — kinesis, kafka, and file — and the same JSON is used by all four languages. See the Streaming guide for context.

kinesis

{
  "type": "kinesis",
  "streamName": "edge-{ThingName}",
  "region": "us-east-1",
  "endpointUrl": "http://localhost:4566"
}

Field	Required	Notes
streamName	yes	Target Kinesis data stream; supports template variables ({ThingName}, {ComponentName}).
region	no	AWS region; falls back to the ambient AWS region when omitted (nullable).
endpointUrl	no	Override the endpoint for LocalStack / floci / a VPC endpoint (nullable).

kafka

{
  "type": "kafka",
  "bootstrapServers": "broker-1:9092,broker-2:9092",
  "topic": "edge.telemetry",
  "properties": { "compression.type": "lz4", "acks": "all" }
}

Field	Required	Notes
bootstrapServers	yes	host:port[,host:port...]; supports template variables.
topic	yes	Target Kafka topic; supports template variables.
properties	no	Open map of librdkafka producer properties, passed through verbatim.

file

Rolling Parquet/AVRO files for later bulk upload to a cloud data lake (S3/Glue/Athena, ADLS, GCS/BigQuery). Built only with the streaming-file-parquet and/or streaming-file-avro cargo feature; absent the feature, a file stream is buffer-only.

{
  "type": "file",
  "format": "parquet",
  "mode": "rows",
  "dir": "/data/{ThingName}/archive",
  "partitionBy": "dt={yyyy-MM-dd}/hr={HH}",
  "maxFileBytes": 134217728,
  "maxFiles": 64,
  "rollEverySecs": 300,
  "onFull": "dropOldest",
  "compression": "snappy"
}

Field	Required	Notes
format	no	parquet (default, columnar/query-ready) or avro (append-friendly, true union value type).
mode	no	rows (default) writes typed columns (see Rows projection below); raw writes one row per message (opaque payload + partitionKey/tsMs/offset).
dir	yes	Output directory root; supports template variables ({ThingName}, {ComponentName}).
partitionBy	no	Hive-style sub-path appended to dir, with UTC time tokens {yyyy} / {MM} / {dd} / {HH} and the compound {yyyy-MM-dd}, resolved per file at roll time.
maxFileBytes	no	Roll a new file once it passes this size (default 134217728 = 128 MiB). Soft cap evaluated at row-group boundaries — a finalized file may exceed it by up to one row group plus the Parquet footer; keep batch.maxBytes well below it for tight files.
maxFiles	no	Keep at most this many finalized files under dir (0 = unbounded).
rollEverySecs	no	Also roll after this many seconds, evaluated on the next write (0 = time-roll disabled).
onFull	no	When maxFiles is reached: dropOldest (default) deletes the oldest finalized file, or stop halts writing.
compression	no	none, snappy (default), zstd, or gzip.
rows	no	Column projection for rows mode (below). Absent → the built-in default projection.

Rows projection

rows columns are declared by config, so you map any message shape to a typed table — SouthboundSignalUpdate is just the default, not a requirement.

• Default projection (when rows is omitted): one row per body.samples[] element, with columns tags (the entire edgecommons envelope tags object as a JSON string), signalId, signalName, adapter, instance, the value in sparse typed columns (valueDouble/valueLong/valueBool/valueString + valueType), quality, qualityRaw, sourceTs, serverTs, tsMs, offset. A payload that isn’t a SouthboundSignalUpdate (not JSON, or no body.samples) goes to a sibling _unmapped raw file rather than being dropped.
• Config-driven projection (rows.columns [+ rows.explode]): you declare the columns and types.

"rows": {
  "explode": "body.readings",
  "columns": [
    { "name": "sensor", "path": "body.id",           "type": "string" },
    { "name": "temp_c", "path": "body.readings[].c", "type": "double" },
    { "name": "ts",     "path": "header.timestamp",  "type": "string" },
    { "name": "tags",   "path": "tags",              "type": "json"   }
  ]
}

Field	Required	Notes
explode	no	Path to an array; emit one row per element. A column path under <explode>[]… resolves against the current element; other paths repeat per row.
columns[].name	yes	Output column name.
columns[].path	yes	Dotted JSON path (body. / tags. / header.; [] under the exploded array).
columns[].type	no	string (default), long, double, bool, or json (serializes objects/arrays). A value that doesn’t match the type is null.

file-sink durability

Delivery is at-least-once. A clean shutdown finalizes the open file (no loss). On a hard crash, Avro recovers up to its last written block while Parquet discards the unclosed, footer-less file (loss bounded by the open-file window — shrink it with rollEverySecs/maxFileBytes). Files are written *.inprogress and atomically renamed on finalize; de-duplicate downstream on (signalId, sourceTs).

callback sinks are programmatic only

The core also supports a callback sink (SinkConfig::Callback) for bring-your-own delivery, but it is not part of the JSON schema — a config with sink.type: "callback" fails schema validation. Callback streams are reachable only programmatically: Java StreamService.registerSink(SinkFunction) (static, before open), Python StreamService.open_with_callback(config, callback), TypeScript registerSinkCallback(streamName, cb) (with the async resolveSinkOutcome, not re-exported from the public entry point), and Rust DefaultStreamService::open_with(config, &SinkFactory). The SinkConfig::Callback variant carries an optional id ({ "type": "callback", "id": "metrics-cw" }) that routes a stream to a specific host callback when several callback streams coexist.

For the full buffer, batch, and delivery config (with defaults and validation), see the Streaming guide and the Configuration reference.

Error type and codes

The error type mirrors the core’s status codes across languages. Java exposes the codes as named constants on EdgeStreamException; Python and TypeScript carry a numeric code; Rust wraps a message string with no numeric code.

Language	Type	Code accessor
Java	EdgeStreamException extends RuntimeException	int code() + constants (ERR_CONFIG, …)
Python	EdgeStreamError(Exception)	e.code (int)
Rust	EdgeCommonsError::Streaming(String) (wraps core edgestreamlog::EdgeStreamError)	none — message string only
TypeScript	EdgeStreamError extends Error	readonly code: number

Code	Name	When
0	OK	success
1	ERR_CONFIG	invalid streaming config / bad JSON
2	ERR_IO	durable-store I/O error
3	ERR_CORRUPT	corrupt on-disk segment or checkpoint
4	ERR_FULL	ingest queue full under onFull: rejectNew
5	ERR_UNKNOWN_STREAM	stream() / stats() for an unconfigured name
6	ERR_SINK	export sink error
7	ERR_PANIC	a Rust panic crossed the native (FFI) boundary
8	ERR_INVALID_ARG	null / invalid native-ABI argument

Java exposes constants 0–8 on EdgeStreamException. Python and TypeScript derive the code from the native message and use a synthetic -1 when the native lib is missing or the message is unparseable. ERR_PANIC (7) and ERR_INVALID_ARG (8) originate in the native layer.

Java

try {
    StreamHandle h = streams.stream("telemetry");
    h.append("pump-7", System.currentTimeMillis(), payload);
} catch (EdgeStreamException e) {
    if (e.code() == EdgeStreamException.ERR_UNKNOWN_STREAM) {
        log.error("no such stream", e);
    } else if (e.code() == EdgeStreamException.ERR_FULL) {
        log.warn("buffer rejected the record (onFull: rejectNew)");
    }
}

Python

from edgecommons.streaming import EdgeStreamError

try:
    h = streams.stream("telemetry")
    h.append("pump-7", int(time.time() * 1000), payload)
except EdgeStreamError as e:
    if e.code == 5:        # ERR_UNKNOWN_STREAM
        log.error("no such stream: %s", e)
    elif e.code == 4:      # ERR_FULL
        log.warning("buffer rejected the record (onFull: rejectNew)")
    # code == -1 means the edgestreamlog_native wheel is missing / unparseable

Rust

// Errors are EdgeCommonsError::Streaming(String) — a message wrapper with no numeric code.
match streams.stream("telemetry") {
    Ok(h) => {
        if let Err(e) = h.append(StreamRecord::new("pump-7", now_ms, payload)) {
            tracing::error!(error = %e, "append failed");
        }
    }
    Err(e) => tracing::error!(error = %e, "stream not configured"),
}

TypeScript

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

try {
  const h = streams.stream("telemetry");
  h.append("pump-7", Date.now(), payload);
} catch (e) {
  if (e instanceof EdgeStreamError) {
    if (e.code === 5) console.error("no such stream");        // ERR_UNKNOWN_STREAM
    else if (e.code === 4) console.warn("buffer rejected");   // ERR_FULL
    // code === -1 means the native addon is missing / message unparseable
  }
}

Native core

The streaming engine lives in the shared edgestreamlog native core. How it is provided, and how you probe its availability, differs per language.

Language	How the core is provided	Probe
Java	cdylib edgestreamlog loaded once per process (-Dedgestreamlog.library.path, then java.library.path, then a bundled classpath resource). Run with --enable-native-access=ALL-UNNAMED.	StreamService.nativeAvailable() (static; returns false instead of throwing)
Python	PyO3 wheel edgestreamlog_native, imported lazily; failure surfaces on first use as EdgeStreamError(-1, ...).	edgecommons.streaming.service.native_available() (not re-exported from streaming/__init__.py)
Rust	Linked directly — availability is compile-time via cargo features.	#[cfg(feature = "streaming")]
TypeScript	napi addon @edgecommons/streamlog-node, require-d lazily on the first StreamService.open.	none — loading is lazy and reported on first use

For Rust, the streaming feature enables the module and accessor; real export also needs a sink feature — streaming-kinesis and/or streaming-kafka. Without a sink feature, a stream still buffers durably but never exports.

Cargo.toml

[dependencies]
edgecommons = { version = "*", features = ["streaming", "streaming-kinesis"] }

The probe call in each language:

Java

if (StreamService.nativeAvailable()) {
    // safe to open streams; returns false instead of throwing if the cdylib is missing
}

Python

from edgecommons.streaming.service import native_available

if native_available():
    ...  # the edgestreamlog_native wheel imported cleanly

Rust

// Availability is compile-time: gate streaming code on the cargo feature.
#[cfg(feature = "streaming")]
{
    // the streaming module + the streams() accessor are compiled in
}

TypeScript

// No nativeAvailable() helper. The @edgecommons/streamlog-node addon is require-d
// lazily on the first StreamService.open(); a EdgeStreamError is thrown if it is missing.

Cross-language differences at a glance

Aspect	Java	Python	Rust	TypeScript
Accessor	getStreams() → null	get_streams() → None	streams() → always Arc<dyn ...>	streams() → StreamService | undefined
append	3 args	3 args	one Record value	3 args
Batch append	—	—	append_batch (Rust-only)	—
Unknown stream	throws	throws	Err / None	throws
Handle close()	yes (AutoCloseable)	yes (context manager)	none (RAII)	yes
Native probe	nativeAvailable()	native_available()	compile-time cfg	none
Stats fields	10	10	11 (last_export_error)	10
BYO-sink	registerSink (static)	open_with_callback	open_with	registerSinkCallback

See also

Streaming guideThe task-oriented walkthrough: enable streaming, buffer/onFull semantics, and delivery.

Configuration reference: streaming sectionEvery streaming.streams[] field, with defaults and validation rules.

Messaging APIThe two-way path — publish/subscribe and request/reply.