Data Streams

Preview API. Data Streams are in Public Preview and may receive breaking changes without notice. You must acknowledge this in code by opting in with @file:OptIn(com.ditto.kotlin.PreviewDataStreams::class).

This guide covers the Android (Kotlin) Data Streams preview and requires Ditto SDK 5.1.0. The code samples here are Kotlin-specific.

What are Data Streams?

A Data Stream is a low-latency, bidirectional, peer-to-peer channel that carries raw bytes between two reachable Ditto peers. Each stream is opened on a named topic — a short label that both peers agree on, similar to a channel name. “Point-to-point” here is logical, not necessarily a single physical hop. NGN resolves a next hop through its router, so a stream between two peers can traverse intermediate peers in the mesh to reach a peer that isn’t directly connected. Data Streams work differently from the document and attachment sync you may already know. Document sync converges structured data through the Ditto store: every peer that subscribes eventually sees the same data, and Ditto persists it for you. Data Streams are the opposite of that model — they carry raw bytes, are ephemeral, and never write anything to the Ditto store. Ditto doesn’t know or care what’s inside the bytes you send; your application defines its own message format on top of the channel, whether that’s Protobuf, JSON, or a custom raw frame format. To open a stream, one peer binds a topic — this makes it act like a server, listening for incoming connections. Another peer then connects to that topic — acting like a client. Once the connection is established, the relationship is symmetric: both sides can send and receive bytes for as long as the stream stays open.

When to use Data Streams

Data Streams and document sync solve different problems, and most apps end up using both. Reach for Data Streams when:

You need real-time, high-frequency, or ephemeral data delivered to another reachable peer, such as sensor readings, telemetry, live cursor or position updates, or audio frames.
The data doesn’t need to be stored, queried, or converged across the mesh — it just needs to get to the other peer, fast.
You want to define your own wire format and manage delivery guarantees yourself.

Prefer document sync when:

The data must be persisted, so peers can read it later, offline or otherwise.
The data must be queried with DQL, or must converge consistently across every peer in the mesh, not just one direct link.

Core concepts

Data Streams introduce a small vocabulary of their own. This glossary covers the terms used throughout the rest of this guide.

Term	Meaning
Peer	One running app instance, identified by a Ditto public peer key.
Presence	Ditto’s view of nearby peers. Used to discover a peer’s key before connecting.
Topic	A short name identifying a kind of stream. Must match `^[a-zA-Z0-9_ ]{1,15}$`.
Acceptor	The handle returned by binding a topic. Keep it alive to keep accepting connections.
Candidate	A callback-scoped opportunity to open a stream. Open it in the callback, or call `take()` to use it later.
Stream	An open, bidirectional byte channel for one topic and reliability mode.
Reliable	Delivered in order, or the connection fails (higher latency).
Unreliable	May be lost, reordered, or duplicated. Lowest latency.
Backpressure	Bounding how much data you queue so a fast sender can’t overwhelm a slow link.

How a stream works, end to end

Before looking at any code, it helps to walk through the full lifecycle of a stream — who does what, and in what order.

The receiving peer binds a topic

The peer that will receive the connection binds a topic and keeps the returned acceptor alive. Dropping the acceptor stops the topic from accepting new connections.

The sending peer discovers the receiver

The peer that will initiate the connection uses presence to discover the receiving peer’s public peer key.

The sender connects on the topic

The sender opens a connection to that peer key on the agreed-upon topic.

The receiver's bind callback fires

On the receiving side, the bind callback fires with a candidate. The candidate is only valid inside the callback, so the app either opens it into a stream immediately, or calls take() to hold onto it and open it later.

Both peers send and receive

Once open, the stream is a symmetric, bidirectional channel. Either peer can send raw byte payloads and receive them from the other side.

Either side closes the stream

When the exchange is finished, either peer can close the stream to release its resources.

Presence only indicates that a peer is nearby — it does not prove that peer has bound the topic yet. Because of this, initial connection attempts can fail while the app is still starting up. Treat connection failures as expected and retry rather than as a fatal error.

Reliable vs. unreliable streams

Every stream is opened as either reliable or unreliable, and both sides of the connection must agree on which one they’re using.

Property	Reliable	Unreliable
Delivery	Delivered in order, or the connection fails	May be lost, reordered, or duplicated
Latency	Higher	Lowest
Compression	Optional LZ4 (`requestCompression`)	Not applicable
Use for	Control messages, RPC, or ordered file chunks	Latency-sensitive data that tolerates loss (e.g. live sensor/audio frames)

Neither mode guarantees once-only delivery, so your application must tolerate a message it has already seen. Choose reliable streams when ordering matters and a gap should break the exchange rather than be silently skipped — for example, control messages, RPC calls, or ordered file chunks. Choose unreliable streams when you’re sending a continuous flow of data where the newest message matters more than every message arriving — for example, sensor or audio frames, where a dropped frame is quickly superseded by the next one.

The connecting peer’s requested reliability mode must match the reliability the topic was bound with. If they don’t match, the connection is rejected.

Prerequisites and setup

Check prerequisites

Ditto SDK 5.1.0
An Android project
Two Android devices or emulators, so you can test a real connection between two peers

Opt in to the preview API

Data Streams are a preview API, so every file that calls into them needs a file-level opt-in:

@file:OptIn(com.ditto.kotlin.PreviewDataStreams::class)

Enable NGN before constructing Ditto

Data Streams require Ditto’s next-generation networking (NGN), which must be enabled before your Ditto instance is constructed. Set both environment variables in Application.onCreate():

import android.system.Os

// Both must be set before DittoFactory.create(...)
Os.setenv("DITTO_NETWORK_ENABLE_NGN", "true", true)
Os.setenv("DITTO_REPLICATION_OVER_NGN", "false", true)

DITTO_NETWORK_ENABLE_NGN=true turns on the NGN transport that Data Streams run over. Data Streams do not work without it.
DITTO_REPLICATION_OVER_NGN=false keeps your regular document and attachment sync unaffected.

For this preview release, we recommend setting DITTO_REPLICATION_OVER_NGN to false.

Start sync

Data Streams need sync running so presence can discover peers. Ditto’s stable peer-to-peer transports are enabled by default, so for Data Streams alone you only need to start sync:

// A TCP listener is NOT required for Data Streams.
// Enable one only if you also need legacy document/attachment sync
// over TCP (for example, connecting to a Big Peer or a fixed-address
// peer). `listen` is advanced configuration and, if set incorrectly,
// can expose your app on the network — leave it off unless you need it.
//
// ditto.updateTransportConfig { config ->
//     config.listen.tcp {
//         enabled = true
//         port = 0 // ephemeral port
//     }
// }

ditto.sync.start()

If you do enable a listener, use updateTransportConfig as shown — it modifies a copy of the transport config already attached to your Ditto instance, so all of your existing legacy transport settings stay intact. Don’t construct a new DittoTransportConfig() from scratch to configure it — doing so disables every transport you don’t explicitly re-enable.

Start sync before you rely on presence to discover peers.

Quickstart: stream sensor readings between two peers

This quickstart builds the two peer roles from the previous section into working code: one peer binds the sensor_data topic and receives readings, the other connects to it and sends them. Both sides use the Reliable reliability mode, and sensor_data satisfies the topic pattern ^[a-zA-Z0-9_ ]{1,15}$.

Receive: own the acceptor and its streams

On the peer that receives data, wrap the acceptor and every open stream in one owner so shutdown can close them deterministically. Closing the acceptor stops new connections but does not close streams that are already open, so the session has to close those itself.

import com.ditto.kotlin.Ditto
import com.ditto.kotlin.DittoAcceptor
import com.ditto.kotlin.DittoReliability
import com.ditto.kotlin.DittoStream
import com.ditto.kotlin.open
import java.util.concurrent.ConcurrentHashMap
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.joinAll
import kotlinx.coroutines.launch

/**
 * Owns the receiving side of the `sensor_data` topic: one acceptor plus every
 * open stream, keyed by peer. Close it once, on shutdown, to tear everything down.
 */
class SensorReceiverSession(
    ditto: Ditto,
    private val onReading: (ByteArray) -> Unit,
) : AutoCloseable {
    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)

    // bindTopic callbacks arrive on their own threads, so this must be thread-safe.
    private val streams = ConcurrentHashMap<String, DittoStream>()

    private val acceptor: DittoAcceptor = ditto.dataStreams.bindTopic(
        topic = "sensor_data",
        reliability = DittoReliability.Reliable,
    ) { candidate ->
        val peer = candidate.peerKeyString()

        // Open inside the callback. The candidate is freed when the callback
        // returns, but the stream is parented to the endpoint and survives.
        val stream = candidate.open { inbound ->
            // Driver thread: hand the owned bytes off without blocking, then return.
            onReading(inbound.payload())
        }

        // Retain the stream; close any previous stream held for this peer.
        streams.put(peer, stream)?.let { old -> scope.launch { old.closeSync() } }

        // Drop the stream from the map once it closes so the map can't grow forever.
        scope.launch {
            stream.waitUntilClosed()
            streams.remove(peer, stream)
        }
    }

    // Teardown runs on `scope`, so close() returns before it finishes. Close
    // the session before you close Ditto, and don't race Ditto teardown against it.
    override fun close() {
        // Stop accepting, then close every stream we still own (unbinding won't).
        acceptor.close()
        val open = streams.values.toList()
        streams.clear()
        scope.launch {
            // Close concurrently so one stuck stream can't starve the rest.
            open.map { launch { it.closeSync() } }.joinAll()
            scope.cancel()
        }
    }
}

The receive callback — the lambda you pass to open (the onReceive parameter) — runs on the connection’s driver thread. Do only non-blocking work (for example, hand the bytes to a queue) and return quickly, or you risk delayed messages and disconnections. The onReading callback above must therefore be non-blocking, such as a Channel.trySend or MutableSharedFlow.tryEmit.

Keep the SensorReceiverSession (and therefore its DittoAcceptor) alive for as long as you want to accept connections. Only one acceptor may bind a given topic at a time — close the existing session before binding the same topic again.

Send: connect once, reuse one stream

On the peer that sends data, connect once and keep the stream open, then reuse it for every reading. Opening a fresh connection per reading defeats the high-frequency use case and invites overlapping connection attempts.

import com.ditto.kotlin.Ditto
import com.ditto.kotlin.DittoReliability
import com.ditto.kotlin.DittoSendStatus
import com.ditto.kotlin.DittoStream
import com.ditto.kotlin.open

/**
 * Owns one outbound `sensor_data` stream and reuses it for many readings.
 * Create it once with [connect], call [send] as often as you like, then close it.
 */
class SensorSenderSession private constructor(
    private val stream: DittoStream,
) : AutoCloseable {
    val maxSendSize: Long = stream.maxSendSize()

    /** Reuses the single open stream — no reconnect per reading. */
    suspend fun send(reading: ByteArray): DittoSendStatus {
        require(reading.size.toLong() <= maxSendSize) { "reading exceeds maxSendSize()" }
        // status is Sent when the payload was sent to the remote peer,
        // not proof the receiver's app processed it.
        return stream.send(reading)
    }

    /**
     * Closes and suspends until the stream is fully closed. Prefer this on
     * shutdown when you want to know the close finished. Both this and [close]
     * flush already-queued sends; only this one waits for closure to complete.
     */
    suspend fun closeAndWait() {
        stream.closeSync()
    }

    /**
     * Non-suspending close for `use { }` / non-coroutine paths. Returns
     * immediately; queued sends are still flushed in the background.
     */
    override fun close() {
        stream.close()
    }

    companion object {
        /** Connects once and holds the stream open. Get [receiverPeerKey] from presence. */
        suspend fun connect(ditto: Ditto, receiverPeerKey: String): SensorSenderSession =
            ditto.dataStreams.connect(
                peer = receiverPeerKey,
                topic = "sensor_data",
                reliability = DittoReliability.Reliable,
                timeoutMs = 10_000,
            ) { candidate ->
                // take() ownership so the stream outlives connect's continuation,
                // then free the candidate (the stream is parented to the endpoint).
                val owned = candidate.take()
                val stream = owned.open()
                owned.close()
                SensorSenderSession(stream)
            }
    }
}

Then connect once and stream readings for as long as the session is open:

import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.withContext

val session = SensorSenderSession.connect(ditto, receiverPeerKey)
try {
    readings.collect { reading -> session.send(reading) }
} finally {
    // closeAndWait() is suspending; run it in NonCancellable so it still
    // completes when this coroutine is being cancelled.
    withContext(NonCancellable) { session.closeAndWait() }
}

connect is suspending and takes a continuation. It waits for and lends an accepted candidate to your block — the candidate is not opened for you. Call candidate.open() in the block, or take() the candidate to open it later. The SDK closes the candidate when your block returns unless you take() it, so take() if you want the stream to outlive the block. Get receiverPeerKey from presence (the receiver’s peerKeyString), and don’t call connect from inside the presence callback itself — see Cancellation and reconnect.

Key patterns

Candidate ownership with `take()`

Candidates are callback-scoped and close when the callback returns. To use one after the callback (e.g. on another coroutine), call take() and close the returned candidate yourself:

ditto.dataStreams.bindTopic("sensor_data") { borrowed ->
    val owned = borrowed.take()
    applicationScope.launch(Dispatchers.IO) {
        owned.use { candidate ->
            candidate.open().use { stream ->
                handleStatus(stream.send(initialPayload))
            }
        }
    }
}

Send modes and `maxSendSize()`

Whatever you send, first make sure the payload fits in a single send. A payload larger than maxSendSize() is rejected, so check it before sending:

require(payload.size.toLong() <= stream.maxSendSize())

There are three ways to send a payload. They are alternatives — pick the one that fits your use case; you do not call all three. Each accepts the same payload: ByteArray.

1. `send` — await delivery status

Use this when you want to know how the send finished. It is a suspending call that returns once the message’s lifecycle ends, giving you a DittoSendStatus (for example Sent, Failed, or Cancelled). This is the default choice when you care whether a message made it out.

val status: DittoSendStatus = stream.send(payload)
// status tells you how the send finished, e.g. DittoSendStatus.Sent

2. `sendAndForget` — fire-and-forget

Use this for high-frequency data where you don’t need per-message status and can tolerate the occasional drop — for example live sensor frames on an unreliable stream. It returns immediately and does not suspend or report a status, so it has the least overhead.

stream.sendAndForget(payload)
// returns immediately; no status is reported

3. `send` with a monitoring block — track progress

Use this when you need to observe the send as it progresses, such as implementing backpressure. You pass a block that receives a DittoSendOperation; the block runs while the send is in flight, and its return value becomes the result of the call.

val terminal: DittoSendStatus = stream.send(payload) { operation ->
    var current = operation.currentStatus()
    while (current == DittoSendStatus.Pending) {
        current = operation.awaitStatusChange()
    }
    current
}

Backpressure

Use the monitoring overload’s awaitStatusChange() to slow a fast producer to what the link can carry. Treat DittoSendStatus.Unknown as an indeterminate terminal result (don’t loop on it), and treat Sent as “sent to the remote peer,” not proof the receiver’s app code processed it.

Closing resources

All Data Streams handles are AutoCloseable. Retain acceptors and streams in a clear owner — like the SensorReceiverSession and SensorSenderSession above — and close them explicitly (or with use). Closing an acceptor unbinds the topic but does not close streams that are already open, so an owner that holds both must close the streams itself. Use closeSync() to stop sends, drain the queue, and wait for closure; waitUntilClosed() only observes closure and can wait indefinitely. On shutdown, cancel jobs, close streams, close acceptors, then close Ditto.

Cancellation and reconnect

Data Streams do not reconnect automatically. If a stream closes — because the peer went away, the link dropped, or a reliable stream failed — it stays closed until your application opens a new one. Build reconnection into your own logic. Preserve coroutine cancellation (rethrow CancellationException) and use bounded exponential backoff with jitter when retrying failed connections. Avoid opening multiple simultaneous connects for the same peer and topic.

Don’t call connect directly inside a presence-observer callback. Copy the peer key out of the observer and launch the connect from a separate coroutine. Presence callbacks are not the place to run a suspending connection, and connecting inline can block presence delivery or overlap with other callbacks for the same peer.

API at a glance

One-line summaries only (link out for full details later):

Type	Purpose
`DittoDataStreamsEndpoint`	Entry point (`ditto.dataStreams`); binds topics and connects to peers.
`DittoAcceptor`	Returned by `bindTopic`; represents an active server binding.
`DittoStreamCandidate`	A callback-scoped opportunity to open a stream; use `open()` or `take()`.
`DittoStream`	An open bidirectional byte channel; `send`, `sendAndForget`, `closeSync`.
`DittoInbound`	A received message; call `payload()` to get an owned `ByteArray`.
`DittoSendOperation`	Monitors/cancels an in-flight send; supports backpressure.
`DittoReliability`	`Reliable` or `Unreliable`.
`DittoStreamStatus`	`Open`, `ClosedByRemote`, `ClosedByLocal`.
`DittoSendStatus`	`Unknown`, `Pending`, `Sent`, `Failed`, `Cancelled`.

AI Agent Prompt

If you use an AI coding assistant, expand and copy the prompt below. It is a self-contained skill for writing correct Ditto Data Streams code against the Kotlin 5.1.0 preview API — the concepts on this page, the ownership and lifecycle rules, and a review checklist, in one block.

Copy AI Data Streams Prompt (Click to Expand)

You are writing Kotlin code that uses the Ditto Data Streams API. Target Ditto SDK 5.1.0, Android/Kotlin only. This API is in PREVIEW.

CONTEXT
Data Streams are low-latency, bidirectional, peer-to-peer byte channels on a named topic. They carry raw bytes, are ephemeral, and NEVER write to the Ditto store — document/attachment sync is a separate mechanism. The application defines its own message format (Protobuf/JSON/raw) on top. One peer BINDS a topic (server role); another CONNECTS to it (client role). Once connected, both sides can send and receive. "Point-to-point" is logical, not one physical hop: NGN resolves a next hop through its router, so a stream can traverse intermediate peers to reach a peer that isn't directly connected.

CONSTRAINTS
- Use only the API verified in the Ditto 5.1.0 source line. Treat it as preview: opt in at file level with `@file:OptIn(com.ditto.kotlin.PreviewDataStreams::class)`. Never combine APIs from different previews; if the project pins a different artifact, inspect that artifact's public declarations before coding.
- This guide/prompt covers Android/Kotlin. Samples are Kotlin-specific.
- A topic must match the regex ^[a-zA-Z0-9_ ]{1,15}$.
- Reliability must match between binder and connector, or the connection is rejected.
- RELIABILITY SEMANTICS: Reliable = delivered in order, or the connection fails (higher latency); use for control messages, RPC, or ordered file chunks. Unreliable = may be lost, reordered, or duplicated (lowest latency); use for loss-tolerant frames. NEITHER mode is once-only — code must tolerate a duplicate message. Do not claim reliable is "guaranteed" or recommend it for "state that must converge" (that is document-sync semantics).
- Only ONE acceptor may bind a given topic at a time. Close the existing acceptor before rebinding the same topic.
- Data Streams do NOT reconnect automatically; a closed stream stays closed until you open a new one.

SETUP (in Application.onCreate(), BEFORE constructing Ditto)
Os.setenv("DITTO_NETWORK_ENABLE_NGN", "true", true)    // required; Data Streams run over NGN
Os.setenv("DITTO_REPLICATION_OVER_NGN", "false", true) // recommended in preview: keep document/attachment sync on existing transports
ditto.sync.start()   // start sync before you rely on presence for peer discovery
A TCP listener is NOT a prerequisite for Data Streams — stable peer-to-peer transports are enabled by default. Enable one ONLY if the app also needs legacy document/attachment sync over TCP. `listen` is advanced config and, set incorrectly, can expose the app on the network. If you do enable it, MODIFY THE CURRENT config (do NOT construct a new DittoTransportConfig() — that disables every transport you don't re-enable):
// ditto.updateTransportConfig { config -> config.listen.tcp { enabled = true; port = 0 } }

WORKFLOW
1. Pick a valid topic and a reliability mode (Reliable = in order or the connection fails; Unreliable = lowest latency, may be lost/reordered/duplicated).
2. Receiver: bindTopic(topic, reliability) { candidate -> ... } and RETAIN the returned DittoAcceptor in an owner (e.g. an AutoCloseable session) for as long as connections should be accepted.
3. Open a candidate inside its callback, or call take() to move it to another coroutine (then close the returned candidate yourself). The stream you open is parented to the endpoint and survives the candidate closing — but you must still retain and close the stream.
4. Sender: obtain the remote peer key from presence, then call the suspending connect(peer, topic, reliability, timeoutMs) { candidate -> candidate.open() ... } from a coroutine. connect does NOT open the candidate for you — it waits for and lends an accepted candidate to your block; your block must open it (call candidate.open() in the block, or take() it and open later); connect closes the candidate when the block returns unless you take() it. Do NOT call connect from inside a presence callback; copy the peer key out and launch connect on a separate coroutine.
5. In onReceive, copy the payload and return fast.
6. Enforce maxSendSize() before sending; pick exactly ONE send mode. Connect ONCE and reuse a single stream for many sends — do not reconnect per message.
7. Close streams and acceptors explicitly (closing an acceptor does NOT close open streams); preserve coroutine cancellation.

API SURFACE (package com.ditto.kotlin)
- ditto.dataStreams : DittoDataStreamsEndpoint — bindTopic(topic, reliability=Reliable, onNewCandidate); suspend connect(peer, topic, requestCompression=false, reliability=Reliable, timeoutMs=0, arguments=null, continuation)
- DittoAcceptor — topic(); reliability()
- DittoStreamCandidate<Args> — peerKeyString(); topic(); arguments(); take(); plus open() extension functions. Candidates from bindTopic are MaySendArguments.Allowed (may pass arguments via open(arguments=...)); candidates from connect are Disallowed.
- DittoStream — suspend send(payload): DittoSendStatus; sendAndForget(payload): Unit; suspend send(payload) { op -> }; maxSendSize(): Long; close(): Unit (non-suspend; returns immediately, queued sends still flush); suspend closeSync(): DittoStreamStatus (suspends until fully closed); suspend waitUntilClosed(): DittoStreamStatus; peerKeyString(); topic()
- DittoInbound — payload(): ByteArray (an owned copy; safe to keep)
- DittoSendOperation — currentStatus(): DittoSendStatus; suspend awaitStatusChange(): DittoSendStatus; cancel()
- enums: DittoReliability { Reliable, Unreliable }; DittoStreamStatus { Open, ClosedByRemote, ClosedByLocal }; DittoSendStatus { Unknown, Pending, Sent, Failed, Cancelled }

RECEIVE (bind — server side). Own the acceptor AND the open streams; closing the acceptor will not close them.
```kotlin
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
private val streams = ConcurrentHashMap<String, DittoStream>()   // bind callbacks are concurrent

val acceptor = ditto.dataStreams.bindTopic(
    topic = "sensor_data",
    reliability = DittoReliability.Reliable,
) { candidate ->
    val peer = candidate.peerKeyString()
    val stream = candidate.open { inbound ->
        // driver thread: copy bytes and hand off without blocking
        channel.trySend(inbound.payload())
    }
    streams.put(peer, stream)?.let { old -> scope.launch { old.closeSync() } }
    scope.launch { stream.waitUntilClosed(); streams.remove(peer, stream) }  // prune closed
}
// retain `acceptor` for the binding's lifetime; on shutdown: acceptor.close() THEN closeSync() each stream
```

SEND (connect — client side). Connect ONCE, keep the stream, reuse it for many sends.
```kotlin
// take() so the stream outlives connect's continuation
val stream: DittoStream = ditto.dataStreams.connect(
    peer = receiverPeerKey,
    topic = "sensor_data",
    reliability = DittoReliability.Reliable,
    timeoutMs = 10_000,
) { candidate ->
    val owned = candidate.take()   // your block MUST open; connect does not
    val s = owned.open()
    owned.close()                  // stream is parented to the endpoint; it survives
    s
}
// then, for each reading:
require(payload.size.toLong() <= stream.maxSendSize())
val status = stream.send(payload)  // pick ONE send mode; reuse `stream` — do not reconnect
// on shutdown: stream.closeSync()
```

SEND MODES (pick exactly one)
- send(payload): DittoSendStatus — suspends until the message lifecycle ends; use when you need the delivery status.
- sendAndForget(payload) — returns immediately, no status; use for high-rate, loss-tolerant data (e.g. Unreliable frames).
- send(payload) { op -> } — monitor via op.currentStatus() / op.awaitStatusChange(); use for backpressure.
Treat DittoSendStatus.Sent as "sent to the remote peer", NOT proof the remote app processed it. Do not loop waiting on Unknown (indeterminate terminal).

OWNERSHIP & LIFECYCLE
- Wrap each role in an AutoCloseable owner (e.g. SensorReceiverSession / SensorSenderSession) that retains its handles and closes them deterministically.
- Candidates from bindTopic/connect are borrowed and close when the callback returns. Open them in-callback, or take() then close the returned candidate. An opened stream survives its candidate closing but must still be retained and closed.
- Closing an acceptor UNBINDS the topic but does NOT close streams already open on it. An owner that holds both must close the streams itself. Prune streams from your map when waitUntilClosed() returns.
- DittoInbound and DittoSendOperation are callback-scoped: copy the payload() bytes; do not retain the wrappers.
- All handles are DittoResource/AutoCloseable. Close with use{} or explicitly. closeSync() initiates local close, drains the queue, and waits; waitUntilClosed() only OBSERVES closure and can wait indefinitely.
- The receive callback runs on the connection's driver thread: do only non-blocking work (Channel.trySend / MutableSharedFlow.tryEmit) and return quickly.
- Shutdown order: cancel connect/retry/monitor jobs -> close streams -> close acceptors -> close Ditto. Use thread-safe collections because bind callbacks can arrive concurrently.

CANCELLATION & RECONNECT
Data Streams do NOT reconnect automatically — build reconnection into your own logic. connect declares CancellationException, but the inspected 5.1.0 implementation can wrap an in-flight cancellation as DittoException.DataStreamsException. Catch CancellationException first and rethrow it; before retrying a DataStreamsException, call currentCoroutineContext().ensureActive() so a cancelled coroutine does not reconnect. Use capped exponential backoff with jitter, and never run multiple simultaneous connects for the same peer + topic. Never call connect from inside a presence-observer callback: copy the peer key out and launch connect on a separate coroutine.

REJECT CODE THAT
- calls connect without its continuation, expects connect to open the candidate for you, or invents a sendSync / treats send(payload) as fire-and-forget
- reconnects per message instead of reusing one open stream for many sends
- returns only the acceptor from a receiver so callers cannot close the open streams, or assumes closing the acceptor closes them
- moves a candidate out of a callback without take(), or fails to close a take()n candidate
- retains DittoInbound or DittoSendOperation outside its callback
- blocks or does heavy work (I/O, decoding, DB, UI) in onReceive
- drops an acceptor or stream reference while it is still needed
- uses waitUntilClosed() as though it initiates closure
- retries after DataStreamsException without checking coroutine activity
- calls connect inside a presence callback, or assumes a closed stream will reconnect on its own
- enables a TCP listener as if it were a Data Streams prerequisite (it is only needed for legacy TCP document sync)
- describes reliable delivery as "guaranteed" / for "state that must converge", or assumes once-only delivery (duplicates are possible)
- replaces the transport config unintentionally, or uses mismatched DITTO_REPLICATION_OVER_NGN across peers that must replicate
- binds the same topic with two acceptors at once
- uses mismatched reliability, an invalid topic, or sends a payload larger than maxSendSize()

Limitations and preview caveats

This guide covers the Android (Kotlin) preview; the samples are Kotlin-specific.
Requires Ditto SDK 5.1.0 and the @file:OptIn(com.ditto.kotlin.PreviewDataStreams::class) annotation.
Topics must match ^[a-zA-Z0-9_ ]{1,15}$.
The receive callback you pass to open runs on the connection’s driver thread and must return quickly.
Streams carry raw bytes and are never persisted to the Ditto store.
We recommend keeping DITTO_REPLICATION_OVER_NGN=false for this preview release.
The API may change during Preview.

Overview & Setup

SDKs

Accessing Data

Syncing Data

Authentication and Authorization

Managing the Mesh

Deploying your App

Data Streams

What are Data Streams?

When to use Data Streams

Core concepts

How a stream works, end to end

Reliable vs. unreliable streams

Prerequisites and setup

Quickstart: stream sensor readings between two peers

Key patterns

Candidate ownership with `take()`

Send modes and `maxSendSize()`

1. `send` — await delivery status

2. `sendAndForget` — fire-and-forget

3. `send` with a monitoring block — track progress

Backpressure

Closing resources

Cancellation and reconnect

API at a glance

AI Agent Prompt

Limitations and preview caveats

Next steps

​What are Data Streams?

​When to use Data Streams

​Core concepts

​How a stream works, end to end

​Reliable vs. unreliable streams

​Prerequisites and setup

​Quickstart: stream sensor readings between two peers

​Key patterns

​Candidate ownership with take()

​Send modes and maxSendSize()

​1. send — await delivery status

​2. sendAndForget — fire-and-forget

​3. send with a monitoring block — track progress

​Backpressure

​Closing resources

​Cancellation and reconnect

​API at a glance

​AI Agent Prompt

​Limitations and preview caveats

​Next steps

What are Data Streams?

When to use Data Streams

Core concepts

How a stream works, end to end

Reliable vs. unreliable streams

Prerequisites and setup

Quickstart: stream sensor readings between two peers

Key patterns

Candidate ownership with `take()`

Send modes and `maxSendSize()`

1. `send` — await delivery status

2. `sendAndForget` — fire-and-forget

3. `send` with a monitoring block — track progress

Backpressure

Closing resources

Cancellation and reconnect

API at a glance

AI Agent Prompt

Limitations and preview caveats

Next steps