Consume live data

Push-based monitoring: a subscription names a set of time-series, and a live connection delivers each new datapoint as it lands — no polling. A control-room dashboard, an alerting worker, or a downstream pipeline drives the loop and acks what it has handled. Anything unacked is redelivered, so a crash never loses data.

1. Create a subscription

Java

Subscription sub = new Subscription();
sub.setExternalId("engine_room");
sub.setName("Engine room");
sub.setTimeseries(List.of(
    IdCollection.createFromExternalId("engine_temperature"),
    IdCollection.createFromExternalId("engine_rpm")));
client.subscriptions().create(List.of(sub));

Python

import datahub_sdk

client.subscriptions.create([
    datahub_sdk.Subscription(
        external_id="engine_room",
        name="Engine room",
        timeseries=["engine_temperature", "engine_rpm"])])

Rust

use dataplatform_rust_sdk::subscriptions::Subscription;
use dataplatform_rust_sdk::generic::IdAndExtId;

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

2. Listen and ack

Hand each message to a handler (or drive a loop yourself), and ack once you've durably handled it. Make the handler idempotent — a redelivery after a crash will replay the last unacked messages.

Java

Register a handler with stream — a dedicated virtual thread delivers each message as it arrives and acks it once your handler returns (or nacks it if the handler throws). The returned handle stops delivery and closes the listener, so use try-with-resources.

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

try (var stream = client.subscriptions().listen(List.of("engine_room"))
        .stream((SubscriptionMessage msg) -> handle(msg.payload()))) {   // auto-acks on return
    awaitShutdown();        // your app's lifecycle; closing the handle ends the stream
}

Prefer a loop, or want to ack on your own schedule? Drive poll yourself — it blocks up to the timeout and returns null on a quiet interval (it is a blocking queue hand-off, not network polling):

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

try (SubscriptionListener listener = client.subscriptions().listen(List.of("engine_room"))) {
    while (running) {
        SubscriptionMessage msg = listener.poll(Duration.ofSeconds(5));
        if (msg == null) continue;
        handle(msg.payload());        // event action + affected datapoints
        listener.ack(msg.messageId());
    }
}

Python

The listener is iterable and a context manager:

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

Rust

next().await yields messages until the socket closes; reconnects are transparent.

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

Change the interest set at runtime

A long-lived listener doesn't need to reconnect to follow more (or fewer) series — subscribe / unsubscribe adjust it in place.

Java

listener.subscribe(List.of("engine_oil_pressure"));
listener.unsubscribe(List.of("engine_rpm"));

Python

listener.subscribe(["engine_oil_pressure"])
listener.unsubscribe(["engine_rpm"])

Rust

listener.subscribe(&["engine_oil_pressure"]).await?;
listener.unsubscribe(&["engine_rpm"]).await?;

nack to retry later

If a message can't be processed right now (a downstream system is down), nack it instead of acking — it will be redelivered rather than dropped. With stream, throwing from the handler nacks for you; to control acks explicitly use stream(handler, AckMode.MANUAL) and call ack/nack yourself. With poll, just call nack instead of ack.