Subscriptions

Durable, fan-out subscriptions over time-series, plus live delivery over a WebSocket.

Manage subscriptions

Java

Subscription sub = new Subscription();
sub.setExternalId("engine_temps");
sub.setName("Engine temps");
sub.setTimeseries(List.of(IdCollection.createFromExternalId("engine_temperature")));
client.subscriptions().create(List.of(sub));

DataWrapper<Subscription> all = client.subscriptions().list(new SubscriptionRetriever());

client.subscriptions().delete(List.of(IdCollection.createFromExternalId("engine_temps")));

Python

import intellistream_datahub_sdk

sub = intellistream_datahub_sdk.Subscription(
    external_id="engine_temps",
    name="Engine temps",
    timeseries=["engine_temperature"])
client.subscriptions.create([sub])

all_subs = client.subscriptions.list()

client.subscriptions.delete(["engine_temps"])

Rust

use intellistream_datahub_sdk::subscriptions::{Subscription, SubscriptionRetriever};
use intellistream_datahub_sdk::generic::IdAndExtId;

let sub = Subscription::new(
    "engine_temps".into(), "Engine temps".into(),
    vec![IdAndExtId::from_external_id("engine_temperature")]);
api.subscriptions.create(&sub).await?;

let all = api.subscriptions.list(&SubscriptionRetriever::default()).await?;

api.subscriptions.delete(&vec![IdAndExtId::from_external_id("engine_temps")]).await?;

Dataset access control

Creating a subscription requires read access to every timeseries' dataset it binds. If you lack read access to any of them, create fails with HTTP 403 and nothing is persisted. Access is granted through Keycloak organization groups: /datasets/<externalId>/read for one data set (and everything beneath it), or the wildcard /datasets/*/read for all of them. Dataset access control →

Live delivery

listen opens an authenticated WebSocket over one or more subscriptions. Stream messages to a handler or drive a loop, and ack the messages you've processed — anything left unacked is redelivered on reconnect.

Java

stream delivers each message on a dedicated virtual thread and auto-acks once the handler returns (throw to nack); the returned handle is AutoCloseable:

import ai.intellistream.datahub.sdk.subscriptions.SubscriptionMessage;

try (var stream = client.subscriptions().listen(List.of("engine_temps"))
        .stream((SubscriptionMessage msg) -> process(msg.payload()))) {
    awaitShutdown();
}

Or drive poll yourself — a blocking queue hand-off (not network polling) that returns null on timeout. Reach for poll, or stream(handler, AckMode.MANUAL), when you need to ack on your own schedule:

import ai.intellistream.datahub.sdk.subscriptions.SubscriptionListener;
import java.time.Duration;

try (SubscriptionListener listener = client.subscriptions().listen(List.of("engine_temps"))) {
    while (running) {
        SubscriptionMessage msg = listener.poll(Duration.ofSeconds(5));
        if (msg == null) continue;
        process(msg.payload());
        listener.ack(msg.messageId());
    }
}

Python

The listener is iterable and a context manager:

with client.subscriptions.listen(["engine_temps"]) as listener:
    for msg in listener:
        process(msg.payload)
        listener.ack([msg.message_id])

Rust

next().await yields Some(Ok(msg)), Some(Err(..)), or None when the socket closes (reconnects are transparent):

let mut listener = api.subscriptions.listen(&["engine_temps"]).await?;
while let Some(result) = listener.next().await {
    match result {
        Ok(msg) => {
            process(&msg.payload);
            listener.ack(&[msg.message_id.as_str()]).await?;
        }
        Err(e) => eprintln!("listen error: {}", e),
    }
}

Every listener also exposes stream for push delivery, ack/nack, subscribe/unsubscribe/set_subscriptions to change the live interest set at runtime, and close.

A delivered message carries the originating subscription's external id, an opaque messageId you echo back to ack/nack, and a payload describing the fan-out event (an action — create/update/delete — plus the affected datapoints).

Refused subscriptions surface as errors

Live delivery enforces the same dataset ACL: to attach a subscription you must be able to read all of its bound timeseries. A subscription you can't read (reason: "forbidden") or one that doesn't exist (reason: "not-found") is refused per-subscription — the connection stays open for the subscriptions that did attach. The refusal is surfaced, not swallowed: a SubscriptionError via pollError in Java, an Err(ListenError::Subscription { .. }) from next().await in Rust, and an exception raised from the iterator in Python — so a refused subscription is visible instead of looking like an indefinitely silent stream.

Acking is at-least-once

Ack a message only after you've durably handled it. If your process dies before the ack, the server redelivers it — so make your handler idempotent.

What each client covers

Operation	Java	Python	Rust
Create	subscriptions().create	subscriptions.create	subscriptions.create
List	subscriptions().list	subscriptions.list	subscriptions.list
Delete	subscriptions().delete	subscriptions.delete	subscriptions.delete
Live delivery	subscriptions().listen	subscriptions.listen	subscriptions.listen

Full parity — subscriptions are the one area where all three clients cover the same ground.