> ## Documentation Index
> Fetch the complete documentation index at: https://docs.ditto.live/llms.txt
> Use this file to discover all available pages before exploring further.

# Data Streams

> Stream low-latency, bidirectional raw bytes between reachable Ditto peers over named topics.

<Warning>
  **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)`.
</Warning>

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

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

<Steps>
  <Step title="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.
  </Step>

  <Step title="The sending peer discovers the receiver">
    The peer that will initiate the connection uses **presence** to discover the receiving peer's public peer key.
  </Step>

  <Step title="The sender connects on the topic">
    The sender opens a connection to that peer key on the agreed-upon topic.
  </Step>

  <Step title="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.
  </Step>

  <Step title="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.
  </Step>

  <Step title="Either side closes the stream">
    When the exchange is finished, either peer can close the stream to release its resources.
  </Step>
</Steps>

<Note>
  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.
</Note>

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

<Warning>
  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.
</Warning>

## Prerequisites and setup

<Steps>
  <Step title="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
  </Step>

  <Step title="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:

    ```kotlin theme={null}
    @file:OptIn(com.ditto.kotlin.PreviewDataStreams::class)
    ```
  </Step>

  <Step title="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()`:

    ```kotlin theme={null}
    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.

    <Note>
      For this preview release, we recommend setting `DITTO_REPLICATION_OVER_NGN` to `false`.
    </Note>
  </Step>

  <Step title="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:

    ```kotlin theme={null}
    // 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()
    ```

    <Warning>
      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.
    </Warning>

    <Note>
      Start sync before you rely on presence to discover peers.
    </Note>
  </Step>
</Steps>

## 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}$`.

<Steps>
  <Step title="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.

    ```kotlin theme={null}
    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()
            }
        }
    }
    ```

    <Warning>
      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`.
    </Warning>

    <Note>
      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.
    </Note>
  </Step>

  <Step title="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.

    ```kotlin theme={null}
    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:

    ```kotlin theme={null}
    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() }
    }
    ```

    <Note>
      `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](#cancellation-and-reconnect).
    </Note>
  </Step>
</Steps>

## 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:

```kotlin theme={null}
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:

```kotlin theme={null}
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.

```kotlin theme={null}
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.

```kotlin theme={null}
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](#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.

```kotlin theme={null}
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.

<Warning>
  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.
</Warning>

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

<Accordion title="Copy AI Data Streams Prompt (Click to Expand)">
  ````text theme={null}
  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()
  ````
</Accordion>

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

## Next steps

* [Syncing Data](/sdk/latest/sync/syncing-data)
* [Customizing Transport Configurations](/sdk/latest/sync/customizing-transport-configurations)
* [Using Mesh Presence](/sdk/latest/sync/using-mesh-presence)


## Related topics

- [Kotlin V4→V5 API Migration Guide](/sdk/latest/migration-guides/kotlin-v4.md)
- [Kotlin Release Notes](/sdk/latest/release-notes/kotlin.md)
- [Change Data Capture](/cloud/cdc.md)