Data Streams

Preview release. Data Streams are in Public Preview. They require Ditto SDK 5.2.0 or later and are available for Swift (iOS and macOS) and Kotlin (Android). The API may receive breaking changes without notice, and you must opt in to it in code — see Opt in to the preview API.

What are Data Streams?

A Data Stream is a low-latency, two-way connection that carries raw bytes from one Ditto peer to another. A peer is any running copy of your app that uses Ditto, such as on a phone, tablet, or laptop. Each stream is opened on a named topic, a short label that both peers agree on, similar to a channel name.

How streams reach other peers

Ditto peers connect to each other over whatever is available nearby, such as Bluetooth Low Energy or a local Wi-Fi network. Together, these connected peers form a mesh: a network in which peers can pass data along for each other, so it isn’t limited to the devices that are directly connected. To learn more, see Mesh Networking. A stream connects two peers even when they aren’t directly connected. For example, if your app on device A can reach device C only through device B, Ditto sends the stream through B for you. Your code names the peer it wants to reach, and Ditto finds the path. The devices in between only relay the bytes. Their apps don’t need to do anything, and unlike multi-hop sync, they don’t store any of the data.

Device A and device C are out of range of each other but both are in range of device B. Bytes travel from A through B to C, and back from C through B to A. B relays the bytes and stores nothing. — Device A connects to device C. They're out of range of each other, so Ditto routes the stream through device B, which passes the bytes along in both directions without storing them.

Data Streams are built on Ditto’s Next Generation Networking (NGN), a new networking layer that finds routes like this across the mesh. You don’t call NGN directly. Its name shows up only in configuration, such as the ngn_data_streams_enabled setting you add during setup. 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.

Two rows compare document sync and Data Streams across devices A, B and C. In the document sync row, a document written on A is copied to B and then C, and each device keeps it in its local store. In the Data Streams row, bytes flow from A through B to C, and every device's store stays empty. — Document sync keeps a copy of the data on every peer that subscribes. Data Streams deliver the bytes without storing them on any peer.

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 table covers the terms used throughout the rest of this guide. For terms used across all of Ditto, see the Glossary.

Term	Meaning
Peer	One running app instance, identified by its 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.

Let the connecting peer send first. When the binding peer opens a stream, the connecting peer may not be ready to receive yet. A message the binding peer sends right after opening its side can be lost, even on a reliable stream. Have the binding peer wait until the connecting peer has sent something before it sends, as the take() example does. The connecting peer can send as soon as connect returns.

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
Payload size (`maxSendSize`)	Large	About one network packet
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.2.0 or later
A Swift (iOS or macOS) or Kotlin (Android) app that already initializes Ditto — see the Swift or Kotlin install guide
Two devices, simulators, 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 must opt in. In Kotlin, add a file-level opt-in annotation. In Swift, import DittoSwift with the PreviewDataStreams SPI — without it, the Data Streams types aren’t visible.

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

@_spi(PreviewDataStreams) import DittoSwift

Turn on Data Streams when Ditto opens

Data Streams are off by default. Turn them on by adding the ngn_data_streams_enabled system parameter to the DittoConfig you already create in your install-guide setup:

val config = DittoConfig(
    databaseId = "REPLACE_ME_WITH_YOUR_DATABASE_ID",
    connect = DittoConfig.Connect.Server(url = "REPLACE_ME_WITH_YOUR_URL"),
    // Data Streams are off by default. This is read once, when Ditto opens.
    systemParameters = mapOf("ngn_data_streams_enabled" to true),
)
val ditto = DittoFactory.create(config)

let config = DittoConfig(
    databaseID: "REPLACE_ME_WITH_YOUR_DATABASE_ID",
    connect: .server(url: URL(string: "REPLACE_ME_WITH_YOUR_URL")!),
    // Data Streams are off by default. This is read once, when Ditto opens.
    systemParameters: ["ngn_data_streams_enabled": .bool(true)]
)
let ditto = try await Ditto.open(config: config)

The parameter is read once, when the Ditto instance opens. Setting it later has no effect.
Document and attachment sync are unaffected: they keep using your existing transports. For this preview release, leave the related ngn_sync_enabled parameter at its default (false).

SDK 5.1 used the DITTO_NETWORK_ENABLE_NGN and DITTO_REPLICATION_OVER_NGN environment variables for this. They still work in 5.2 but are deprecated; use the system parameter instead.

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 (after the authentication setup from the install guide):

// 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
//     config.listen.tcp.port = 0 // ephemeral port
// }

ditto.sync.start()

// 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 in
//     config.listen.tcp.isEnabled = true
//     config.listen.tcp.port = 0 // ephemeral port
// }

try 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 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()
        // Remove each stream as we take it: the cleanup coroutines above may be
        // removing streams at the same time, and remove() is atomic per key.
        val open = streams.keys.mapNotNull { streams.remove(it) }
        scope.launch {
            // Close concurrently so one stuck stream can't starve the rest.
            open.map { launch { it.closeSync() } }.joinAll()
            scope.cancel()
        }
    }
}

@_spi(PreviewDataStreams) import DittoSwift
import Foundation

/// Owns the receiving side of the `sensor_data` topic: one acceptor plus every
/// open stream, keyed by peer. Call `close()` once, on shutdown, to tear everything down.
final class SensorReceiverSession: @unchecked Sendable {
    private let onReading: @Sendable (Data) -> Void

    // bindTopic callbacks arrive on their own threads, so guard shared state with a lock.
    private let lock = NSLock()
    private var acceptor: DittoAcceptor?
    private var streams: [String: DittoStream] = [:]

    init(ditto: Ditto, onReading: @escaping @Sendable (Data) -> Void) throws {
        self.onReading = onReading
        let acceptor = try ditto.dataStreams.bindTopic("sensor_data", reliability: .reliable) { [weak self] candidate in
            self?.accept(candidate)
        }
        locked { self.acceptor = acceptor }
    }

    private func accept(_ candidate: DittoStreamCandidate<MaySendArguments.Allowed>) {
        do {
            let peer = try candidate.peerKey

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

            // Retain the stream — ARC closes a stream as soon as nothing references
            // it. Close any previous stream held for this peer.
            let previous = locked { streams.updateValue(stream, forKey: peer) }
            previous?.close()

            // Drop the stream once it closes so the dictionary can't grow forever.
            Task { [weak self] in
                _ = try? await stream.waitUntilClosed()
                self?.locked {
                    if self?.streams[peer] === stream { self?.streams[peer] = nil }
                }
            }
        } catch {
            print("Failed to open sensor_data stream: \(error)")
        }
    }

    /// Stops accepting, then closes every stream this session still owns
    /// (closing the acceptor does not close them). Queued sends still flush.
    func close() {
        let (acceptor, open) = locked {
            defer { self.acceptor = nil; streams.removeAll() }
            return (self.acceptor, Array(streams.values))
        }
        acceptor?.close()
        for stream in open {
            stream.close()
        }
    }

    private func locked<T>(_ body: () throws -> T) rethrows -> T {
        lock.lock()
        defer { lock.unlock() }
        return try body()
    }
}

The receive callback — the closure you pass to open (the onReceive parameter) — runs on the connection’s driver thread. Do only non-blocking work and return quickly, or you risk delayed messages and disconnections. The onReading callback above must therefore be non-blocking: for example Channel.trySend or MutableSharedFlow.tryEmit in Kotlin, or an AsyncStream continuation’s yield in Swift.

Swift: keep a strong reference to every stream you open, as the session above does. A DittoStream that nothing references is released immediately and stops delivering messages.

Keep the SensorReceiverSession (and therefore its acceptor) 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)
            }
    }
}

@_spi(PreviewDataStreams) import DittoSwift
import Foundation

/// Owns one outbound `sensor_data` stream and reuses it for many readings.
/// Create it once with `connect(ditto:receiverPeerKey:)`, call `send(_:)` as often
/// as you like, then close it.
final class SensorSenderSession: Sendable {
    enum SendError: Error {
        case readingTooLarge(size: Int, maxSendSize: Int)
    }

    private let stream: DittoStream
    let maxSendSize: Int

    private init(stream: DittoStream) throws {
        self.stream = stream
        self.maxSendSize = try stream.maxSendSize
    }

    /// Reuses the single open stream — no reconnect per reading.
    func send(_ reading: Data) async throws -> DittoSendStatus {
        guard reading.count <= maxSendSize else {
            throw SendError.readingTooLarge(size: reading.count, maxSendSize: maxSendSize)
        }
        // status is .sent when the payload was sent to the remote peer,
        // not proof the receiver's app processed it.
        return try await stream.send(reading)
    }

    /// Closes and waits 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.
    func closeAndWait() async throws {
        _ = try await stream.closeAndWait()
    }

    /// Returns immediately; queued sends are still flushed in the background.
    func close() {
        stream.close()
    }

    /// Connects once and holds the stream open. Get `receiverPeerKey` from presence.
    static func connect(ditto: Ditto, receiverPeerKey: String) async throws -> SensorSenderSession {
        try await ditto.dataStreams.connect(
            peerKey: receiverPeerKey,
            topic: "sensor_data",
            reliability: .reliable,
            timeoutMs: 10_000
        ) { candidate in
            // The SDK closes the candidate when this closure returns. The stream
            // opened from it is parented to the endpoint and stays open.
            try SensorSenderSession(stream: candidate.open())
        }
    }
}

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() }
}

let session = try await SensorSenderSession.connect(ditto: ditto, receiverPeerKey: receiverPeerKey)
do {
    for await reading in readings {
        _ = try await session.send(reading)
    }
} catch {
    // Close before rethrowing so a failed send doesn't leak the stream.
    try? await session.closeAndWait()
    throw error
}
try await session.closeAndWait()

connect waits for and lends an accepted candidate to your closure — the candidate is not opened for you, so your closure must open it. The SDK closes the candidate when the closure returns; the stream you opened from it stays open. (In Kotlin, call take() first if you want to keep the candidate itself beyond the closure.) Get receiverPeerKey from presence (the receiver’s peer key), 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 (for example, on another coroutine or task), call take() and close the returned candidate yourself:

val acceptor = ditto.dataStreams.bindTopic("sensor_data") { borrowed ->
    val owned = borrowed.take()
    applicationScope.launch(Dispatchers.IO) {
        owned.use { candidate ->
            val requests = Channel<ByteArray>(Channel.UNLIMITED)
            candidate.open { inbound -> requests.trySend(inbound.payload()) }.use { stream ->
                // Reply once the connecting peer has spoken. A message sent right
                // after accepting can arrive before that peer is ready, and be lost.
                requests.receive()
                handleStatus(stream.send(reply))
            }
        }
    }
}

let acceptor = try ditto.dataStreams.bindTopic("sensor_data") { borrowed in
    // take() moves ownership out of the callback; you must close it yourself.
    let candidate = borrowed.take()
    Task {
        defer { candidate.close() }
        do {
            let (requests, continuation) = AsyncStream.makeStream(of: Data.self)
            let stream = try candidate.open { inbound in
                if let payload = try? inbound.payload { continuation.yield(payload) }
            }
            defer { stream.close() }
            // Reply once the connecting peer has spoken. A message sent right
            // after accepting can arrive before that peer is ready, and be lost.
            for await _ in requests.prefix(1) {
                handleStatus(try await stream.send(reply))
            }
        } catch {
            print("Failed to reply on sensor_data: \(error)")
        }
    }
}

The binding peer replies only after the connecting peer’s first message arrives. Sending right after opening the stream is a race: the message can arrive before the connecting peer is ready and be lost (see How a stream works).

Send modes and `maxSendSize`

Whatever you send, first make sure the payload fits in a single send. A payload larger than the stream’s maxSendSize is rejected, so check it before sending. Unreliable streams carry each payload in a single network packet, so their limit is small (on the order of a kilobyte); reliable streams accept far larger payloads.

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

guard payload.count <= (try stream.maxSendSize) else {
    throw DataStreamsExampleError.payloadTooLarge
}

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 in Kotlin, Data in Swift).

1. `send` — await delivery status

Use this when you want to know how the send finished. It suspends (Kotlin) or awaits (Swift) until 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

let status: DittoSendStatus = try await stream.send(payload)
// status tells you how the send finished, e.g. .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 report a status, so it has the least overhead.

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

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

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

Use this when you need to observe the send as it progresses, such as implementing backpressure. You pass a closure that receives a DittoSendOperation; the closure 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
}

let terminal: DittoSendStatus = try await stream.send(payload) { operation in
    var current = try operation.currentStatus
    while current == .pending {
        current = try await operation.awaitStatusChange()
    }
    return current
}

Backpressure

Use the monitoring overload’s awaitStatusChange() to slow a fast producer to what the link can carry. Treat the unknown status 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

Every Data Streams handle — acceptors, candidates, and streams — has a close() method (and in Kotlin is AutoCloseable, so use { } works). Retain acceptors and streams in a clear owner — like the SensorReceiverSession and SensorSenderSession above — and close them explicitly. 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. To stop sends, drain the queue, and wait for closure, use closeSync() in Kotlin or closeAndWait() in Swift. waitUntilClosed() only observes closure and can wait indefinitely. On shutdown, cancel jobs or tasks, close streams, close acceptors, then close (Kotlin) or release (Swift) 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 cancellation — rethrow CancellationException in Kotlin, and stop retrying when Task.isCancelled is true in Swift — and use bounded exponential backoff with jitter when retrying failed connections. An in-flight cancellation can surface as a Data Streams error rather than a cancellation error, so check for cancellation before you retry. 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 start the connect from a separate coroutine or task. Presence callbacks are not the place to run an asynchronous connection, and connecting inline can block presence delivery or overlap with other callbacks for the same peer.

API at a glance

The two SDKs share the same types. Kotlin exposes most accessors as functions, Swift as (throwing) properties.

Type	Purpose	Kotlin	Swift
`DittoDataStreamsEndpoint`	Entry point (`ditto.dataStreams`); binds topics and connects to peers.	`bindTopic(topic, reliability) { }`, `connect(peer = …) { }`	`bindTopic(_:reliability:onNewCandidate:)`, `connect(peerKey:…) { }`
`DittoAcceptor`	Returned by `bindTopic`; represents an active server binding.	`topic()`, `reliability()`	`topic`, `reliability`
`DittoStreamCandidate`	A callback-scoped opportunity to open a stream; use `open()` or `take()`.	`peerKeyString()`	`peerKey`
`DittoStream`	An open bidirectional byte channel; `send`, `sendAndForget`, close.	`maxSendSize()`, `closeSync()`	`maxSendSize`, `closeAndWait()`
`DittoInbound`	A received message; read its payload.	`payload()` → `ByteArray`	`payload` → `Data`
`DittoSendOperation`	Monitors/cancels an in-flight send; supports backpressure.	`currentStatus()`, `awaitStatusChange()`	`currentStatus`, `awaitStatusChange()`
`DittoReliability`	Reliable or unreliable.	`Reliable`, `Unreliable`	`.reliable`, `.unreliable`
`DittoStreamStatus`	How a stream ended.	`Open`, `ClosedByRemote`, `ClosedByLocal`	`.open`, `.closedByRemote`, `.closedByLocal`
`DittoSendStatus`	How a send ended.	`Unknown`, `Pending`, `Sent`, `Failed`, `Cancelled`	`.unknown`, `.pending`, `.sent`, `.failed`, `.cancelled`

AI Agent Prompts

If you use an AI coding assistant, expand and copy the prompt for your platform. Each is a self-contained skill for writing correct Ditto Data Streams code against the 5.2.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 for Kotlin (Click to Expand)

You are writing Kotlin (Android) code that uses the Ditto Data Streams API. Target Ditto SDK 5.2.0 or later. 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. Peers do not need a direct connection: Ditto's Next Generation Networking (NGN) layer routes a stream through intermediate peers in the mesh when needed.

CONSTRAINTS
- Opt in at file level in every file that uses the API: `@file:OptIn(com.ditto.kotlin.PreviewDataStreams::class)`. If the project pins a different SDK artifact, inspect that artifact's public declarations before coding.
- 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).
- Unreliable streams carry one network packet per send, so their maxSendSize() is small (~1 KB); reliable streams allow far larger payloads.
- 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 (when constructing Ditto)
val config = DittoConfig(
    databaseId = ..., connect = ...,                                  // the app's existing config
    systemParameters = mapOf("ngn_data_streams_enabled" to true),     // required; read once at open
)
val ditto = DittoFactory.create(config)
ditto.sync.start()   // start sync before you rely on presence for peer discovery
Leave `ngn_sync_enabled` at its default (false) so document/attachment sync stays on existing transports. The 5.1 env vars DITTO_NETWORK_ENABLE_NGN / DITTO_REPLICATION_OVER_NGN are deprecated; prefer the system parameter.
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; config.listen.tcp.port = 0 }

WORKFLOW
1. Pick a valid topic and a reliability mode.
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 — your block must open it; 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(); close()
- DittoStreamCandidate<Args> — peerKeyString(); topic(); arguments(); take(); close(); 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 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.
- The CONNECTING peer sends first. A message the binding (accepting) peer sends right after open() can arrive before the connecting peer is ready and be lost, even on a Reliable stream. The binding peer must wait for the connecting peer's first message before sending.
- 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. An in-flight cancellation of connect can surface as DittoException.DataStreamsException rather than CancellationException. 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
- has the binding (accepting) peer send right after opening a stream, before the connecting peer has sent anything
- 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)
- turns Data Streams on with Os.setenv instead of the ngn_data_streams_enabled system parameter, or sets ngn_sync_enabled
- describes reliable delivery as "guaranteed" / for "state that must converge", or assumes once-only delivery (duplicates are possible)
- replaces the transport config unintentionally
- binds the same topic with two acceptors at once
- uses mismatched reliability, an invalid topic, or sends a payload larger than maxSendSize()

Copy AI Data Streams Prompt for Swift (Click to Expand)

You are writing Swift (iOS/macOS) code that uses the Ditto Data Streams API. Target Ditto SDK 5.2.0 or later (DittoSwiftPackage). 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. Peers do not need a direct connection: Ditto's Next Generation Networking (NGN) layer routes a stream through intermediate peers in the mesh when needed.

CONSTRAINTS
- Opt in in every file that uses the API: `@_spi(PreviewDataStreams) import DittoSwift`. Without it the Data Streams types are not visible. SPI types cannot appear in a module's `public` API, so keep Data Streams code internal to the app target (or mark wrappers @_spi too).
- 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).
- Unreliable streams carry one network packet per send, so their maxSendSize is small (~1 KB); reliable streams allow far larger payloads.
- 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 (when opening Ditto)
let config = DittoConfig(
    databaseID: ..., connect: ...,                                    // the app's existing config
    systemParameters: ["ngn_data_streams_enabled": .bool(true)]       // required; read once at open
)
let ditto = try await Ditto.open(config: config)
try ditto.sync.start()   // start sync before you rely on presence for peer discovery
Leave `ngn_sync_enabled` at its default (false) so document/attachment sync stays on existing transports.
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, and then MODIFY THE CURRENT config with ditto.updateTransportConfig { config in ... } — never assign a fresh DittoTransportConfig(), which disables every transport you don't re-enable.

WORKFLOW
1. Pick a valid topic and a reliability mode.
2. Receiver: try ditto.dataStreams.bindTopic(topic, reliability:) { candidate in ... } and RETAIN the returned DittoAcceptor in an owner for as long as connections should be accepted.
3. Open a candidate inside its callback, or call take() to move it to another Task (then close the returned candidate yourself). The stream you open survives the candidate closing — but you MUST keep a strong reference to it: a DittoStream that nothing references is released immediately and stops delivering messages.
4. Sender: obtain the remote peer key from presence, then `try await ditto.dataStreams.connect(peerKey:topic:reliability:timeoutMs:) { candidate in try candidate.open() }`. connect does NOT open the candidate for you — your closure must open it; the SDK closes the candidate when the closure returns, and the stream you opened stays open. Return the stream (or an owner wrapping it) from the closure. Do NOT call connect from inside a presence callback; copy the peer key out and start connect from a separate Task.
5. In onReceive, read `inbound.payload` (a Data copy) 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); respect Task cancellation.

API SURFACE (module DittoSwift, SPI PreviewDataStreams)
- ditto.dataStreams : DittoDataStreamsEndpoint — bindTopic(_ topic:, reliability: = .reliable, onNewCandidate: @Sendable (DittoStreamCandidate<MaySendArguments.Allowed>) -> Void) throws -> DittoAcceptor; connect<T>(peerKey:, topic:, requestCompression: = false, reliability: = .reliable, timeoutMs: UInt32 = 0, arguments: Data? = nil, continuation: (DittoStreamCandidate<MaySendArguments.Disallowed>) async throws -> T) async throws -> T
- DittoAcceptor — topic / reliability (throwing properties); close()
- DittoStreamCandidate<Args> — peerKey / topic / arguments (throwing properties); take(); close(); open() / open(onReceive:) (bind candidates also accept arguments:)
- DittoStream — send(_ payload: Data) async throws -> DittoSendStatus; sendAndForget(_:) throws; send(_:withSendOperation:) async throws -> T; maxSendSize: Int (throwing); close() (returns immediately, queued sends still flush); closeAndWait() async throws -> DittoStreamStatus; waitUntilClosed() async throws -> DittoStreamStatus; peerKey; topic
- DittoInbound — payload: Data (throwing property; an owned copy)
- DittoSendOperation — currentStatus (throwing property); awaitStatusChange() async throws -> DittoSendStatus; cancel() throws
- enums: DittoReliability { .reliable, .unreliable }; DittoStreamStatus { .open, .closedByRemote, .closedByLocal }; DittoSendStatus { .unknown, .pending, .sent, .failed, .cancelled }
- Errors are thrown as DittoError (e.g. .dataStreamsError(message:)).

RECEIVE (bind — server side). Own the acceptor AND the open streams; closing the acceptor will not close them.
```swift
final class Receiver: @unchecked Sendable {
    private let lock = NSLock()                     // bind callbacks are concurrent
    private var streams: [String: DittoStream] = [:]
    private var acceptor: DittoAcceptor?

    init(ditto: Ditto, onPayload: @escaping @Sendable (Data) -> Void) throws {
        let acceptor = try ditto.dataStreams.bindTopic("sensor_data", reliability: .reliable) { [weak self] candidate in
            guard let self, let peer = try? candidate.peerKey,
                  let stream = try? candidate.open(onReceive: { inbound in
                      if let payload = try? inbound.payload { onPayload(payload) }  // driver thread: return fast
                  }) else { return }
            self.lock.lock(); let old = self.streams.updateValue(stream, forKey: peer); self.lock.unlock()
            old?.close()                            // retain the new stream; close the replaced one
        }
        lock.lock(); self.acceptor = acceptor; lock.unlock()
    }
    // on shutdown: acceptor.close() THEN close() each stream
}
```

SEND (connect — client side). Connect ONCE, keep the stream, reuse it for many sends.
```swift
let stream = try await ditto.dataStreams.connect(
    peerKey: receiverPeerKey, topic: "sensor_data", reliability: .reliable, timeoutMs: 10_000
) { candidate in
    try candidate.open()          // your closure MUST open; connect does not
}
// keep `stream` alive (store it); then, for each reading:
guard payload.count <= (try stream.maxSendSize) else { throw MyError.tooLarge }
let status = try await stream.send(payload)   // pick ONE send mode; reuse `stream`
// on shutdown: _ = try await stream.closeAndWait()
```

SEND MODES (pick exactly one)
- send(_:) async -> DittoSendStatus — awaits the message lifecycle; use when you need the delivery status.
- sendAndForget(_:) — returns immediately, no status; use for high-rate, loss-tolerant data (e.g. .unreliable frames).
- send(_:) { operation in ... } — monitor via operation.currentStatus / operation.awaitStatusChange(); use for backpressure.
Treat .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 owner class that retains its handles (acceptor + every stream) and closes them deterministically. Guard shared state with a lock; bind callbacks arrive on arbitrary threads and closures must be @Sendable.
- Candidates from bindTopic/connect are borrowed and close when the callback returns. Open them in-callback, or take() then close the returned candidate.
- Closing an acceptor UNBINDS the topic but does NOT close streams already open on it. Prune streams when waitUntilClosed() returns.
- DittoInbound and DittoSendOperation are callback-scoped: copy payload; do not retain the wrappers.
- closeAndWait() 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 (e.g. AsyncStream.Continuation.yield) and return quickly.
- The CONNECTING peer sends first. A message the binding (accepting) peer sends right after open() can arrive before the connecting peer is ready and be lost, even on a .reliable stream. The binding peer must wait for the connecting peer's first message before sending.
- Shutdown order: cancel connect/retry/monitor Tasks -> close streams -> close acceptors -> stop sync / release Ditto.

CANCELLATION & RECONNECT
Data Streams do NOT reconnect automatically — build reconnection into your own logic. An in-flight cancellation can surface as a DittoError rather than CancellationError: catch CancellationError and stop; before retrying any other error, check Task.isCancelled. 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.

REJECT CODE THAT
- forgets `@_spi(PreviewDataStreams)` on the DittoSwift import, or exposes SPI types in public API
- discards the DittoStream returned by open() (it is released immediately) or drops the acceptor while still needed
- expects connect to open the candidate for you, or reconnects per message instead of reusing one open stream
- assumes closing the acceptor closes open streams
- moves a candidate out of a callback without take(), or fails to close a take()n candidate
- has the binding (accepting) peer send right after opening a stream, before the connecting peer has sent anything
- retains DittoInbound or DittoSendOperation outside its callback
- blocks or does heavy work (I/O, decoding, DB, UI) in onReceive
- uses waitUntilClosed() as though it initiates closure
- retries after an error without checking Task.isCancelled
- 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, or replaces the transport config
- turns Data Streams on any way other than the ngn_data_streams_enabled system parameter at open, or sets ngn_sync_enabled
- describes reliable delivery as "guaranteed" / for "state that must converge", or assumes once-only delivery (duplicates are possible)
- 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

Requires Ditto SDK 5.2.0 or later, on Swift (iOS and macOS) or Kotlin (Android).
You must opt in to the preview API: @file:OptIn(com.ditto.kotlin.PreviewDataStreams::class) in Kotlin, @_spi(PreviewDataStreams) import DittoSwift in Swift.
Data Streams must be turned on with the ngn_data_streams_enabled system parameter when Ditto opens.
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.
A message the binding peer sends right after opening a stream can be lost, because the connecting peer may not be ready yet. Let the connecting peer send first.
Streams carry raw bytes and are never persisted to the Ditto store.
We recommend leaving ngn_sync_enabled at its default (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?

How streams reach other peers

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 closure — track progress

Backpressure

Closing resources

Cancellation and reconnect

API at a glance

AI Agent Prompts

Limitations and preview caveats

Next steps

​What are Data Streams?

​How streams reach other peers

​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 closure — track progress

​Backpressure

​Closing resources

​Cancellation and reconnect

​API at a glance

​AI Agent Prompts

​Limitations and preview caveats

​Next steps

What are Data Streams?

How streams reach other peers

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 closure — track progress

Backpressure

Closing resources

Cancellation and reconnect

API at a glance

AI Agent Prompts

Limitations and preview caveats

Next steps