---
name: dart-libp2p-custom-protocol
description: Use when writing your own libp2p protocol in Dart on a dart_libp2p host: protocol IDs, setStreamHandler, newStream, varint length-prefixed framing, P2PStream read/write, deadlines and timeouts, closeWrite vs close vs reset, versioning, request/response and long-lived streams, go-libp2p interop.
triggers: setStreamHandler, newStream, P2PStream, StreamHandler, ProtocolID, setStreamHandlerMatch, closeWrite, encodeVarint, BufferedP2PStreamReader, YamuxStreamTimeoutException, dart_libp2p protocol
---

# Custom protocols on dart_libp2p

## What it is

A libp2p protocol is a protocol ID (a string such as `/myapp/echo/1.0.0`)
plus the bytes two peers exchange on a stream opened for it. The host
negotiates the ID with multistream-select, then hands you a `P2PStream`: an
ordered, reliable, flow-controlled byte stream (a Yamux stream inside a Noise
session). You define the framing and the message flow.

Use something else for: broadcast to many peers (`dart-libp2p-gossipsub`),
key/value lookup (`dart-libp2p-kad-dht`), block exchange
(`dart-libp2p-bitswap`). Building the host: `dart-libp2p-host`.

## Install

```yaml
dependencies:
  dart_libp2p: ^4.12.0
```

```dart
import 'dart:async';
import 'dart:typed_data';
import 'package:dart_libp2p/dart_libp2p.dart' hide Option; // Host, P2PStream, PeerId, Context
import 'package:dart_libp2p/utils/varint.dart';              // encodeVarint, decodeVarint
```

## Core workflow

### Echo: the minimal pair

```dart
const echoId = '/myapp/echo/1.0.0';

// Server: register before host.start() so the first Identify lists it.
host.setStreamHandler(echoId, (P2PStream stream, PeerId remote) async {
  final data = await readAll(stream, maxBytes: 64 * 1024);
  await stream.write(data);
  await stream.closeWrite();      // FIN: tells the client the reply is complete
});

// Client
final s = await host.newStream(peer, [echoId], Context());
await s.write(Uint8List.fromList([1, 2, 3]));
await s.closeWrite();             // half-close: "request done", keep reading
final reply = await readAll(s, maxBytes: 64 * 1024);
await s.close();
```

`read()` returns whatever chunk arrived, of any size, and an **empty list at
end of stream** (the peer's FIN or a reset). Never assume one `read()` is one
message.

```dart
/// Reads until EOF, refusing more than [maxBytes].
Future<Uint8List> readAll(P2PStream s, {required int maxBytes}) async {
  final out = BytesBuilder(copy: false);
  while (true) {
    final chunk = await s.read();
    if (chunk.isEmpty) return out.takeBytes();
    if (out.length + chunk.length > maxBytes) {
      await s.reset();
      throw StateError('message too large');
    }
    out.add(chunk);
  }
}
```

### Length-prefixed framing (go-libp2p compatible)

For more than one message per stream, prefix each message with its length as
an unsigned varint. This is what go-libp2p's `msgio` varint reader and
`pbio.NewDelimitedReader` expect, and what dart_libp2p itself uses for DCUtR,
Circuit Relay v2 and AutoNAT v2.

```dart
Future<void> writeFrame(P2PStream s, Uint8List msg) async {
  final b = BytesBuilder(copy: false)..add(encodeVarint(msg.length))..add(msg);
  await s.write(b.takeBytes());   // one write: prefix and body together
}

class FrameReader {
  FrameReader(this._s, {this.maxFrame = 1 << 20});
  final P2PStream _s;
  final int maxFrame;
  final _buf = BytesBuilder(copy: false);
  Uint8List _pending = Uint8List(0);
  int _pos = 0;

  int get _available => _pending.length - _pos;

  Future<bool> _fill() async {
    final chunk = await _s.read();
    if (chunk.isEmpty) return false;              // EOF
    _buf..add(Uint8List.sublistView(_pending, _pos))..add(chunk);
    _pending = _buf.takeBytes();
    _pos = 0;
    return true;
  }

  /// Next message, or null at a clean EOF between messages.
  Future<Uint8List?> next() async {
    var len = 0, shift = 0;
    for (var i = 0;; i++) {
      if (_available == 0 && !await _fill()) {
        if (i == 0) return null;
        throw const FormatException('EOF inside length prefix');
      }
      final byte = _pending[_pos++];
      len |= (byte & 0x7f) << shift;
      if (byte & 0x80 == 0) break;
      shift += 7;
      if (i >= 8) throw const FormatException('varint too long');
    }
    if (len > maxFrame) throw FormatException('frame of $len bytes');
    while (_available < len) {
      if (!await _fill()) throw const FormatException('EOF inside frame');
    }
    final msg = Uint8List.sublistView(_pending, _pos, _pos + len);
    _pos += len;
    return Uint8List.fromList(msg);
  }
}
```

The package also ships `BufferedP2PStreamReader`
(`package:dart_libp2p/p2p/protocol/circuitv2/util/buffered_reader.dart`) with
`readVarint()`, `readExact(n)` and `isEOF`; it throws on EOF mid-read.

For protobuf messages, write `msg.writeToBuffer()` as the frame body and parse
with `MyMsg.fromBuffer(frame)`.

### Request/response with a timeout

One stream per request is the libp2p norm: it is cheap (Yamux) and the
stream end delimits the reply.

```dart
Future<Uint8List> request(Host host, PeerId peer, Uint8List req) async {
  final s = await host.newStream(peer, ['/myapp/kv/1.0.0'], Context())
      .timeout(const Duration(seconds: 15));
  try {
    await s.setDeadline(DateTime.now().add(const Duration(seconds: 10)));
    await writeFrame(s, req).timeout(const Duration(seconds: 10));
    await s.closeWrite();
    final reply = await FrameReader(s).next().timeout(const Duration(seconds: 10));
    if (reply == null) throw StateError('peer closed without a reply');
    await s.close();
    return reply;
  } catch (_) {
    await s.reset();     // abort: the peer sees a reset, resources are freed
    rethrow;
  }
}

host.setStreamHandler('/myapp/kv/1.0.0', (s, remote) async {
  await s.setDeadline(DateTime.now().add(const Duration(seconds: 10)));
  final req = await FrameReader(s, maxFrame: 64 * 1024).next();
  if (req == null) { await s.close(); return; }
  await writeFrame(s, handle(remote, req));
  await s.closeWrite();
  await s.close();
});
```

### Long-lived stream

```dart
host.setStreamHandler('/myapp/feed/1.0.0', (s, remote) async {
  final reader = FrameReader(s);
  try {
    while (true) {
      final msg = await reader.next();
      if (msg == null) break;          // peer closed its side
      onMessage(remote, msg);
    }
  } finally {
    await s.close();
  }
});
```

Without a deadline a Yamux read waits at most 5 minutes, then throws
`YamuxStreamTimeoutException`. On an idle long-lived stream send an
application heartbeat frame more often than that, or catch the timeout and
keep reading. Watch `host.eventBus` for `EvtPeerConnectednessChanged`
(`notConnected`) and reopen the stream when the peer reconnects.

### Versioning

```dart
// Server speaks both; the negotiated ID tells you which.
for (final id in ['/myapp/kv/2.0.0', '/myapp/kv/1.0.0']) {
  host.setStreamHandler(id, (s, remote) async {
    if (s.protocol() == '/myapp/kv/2.0.0') return serveV2(s, remote);
    return serveV1(s, remote);
  });
}

// Client lists preferences in order; multistream picks the first the peer has.
final s = await host.newStream(peer, ['/myapp/kv/2.0.0', '/myapp/kv/1.0.0'], Context());
print(s.protocol());

// Or match a family with one handler:
host.setStreamHandlerMatch('/myapp/kv/', (id) => id.startsWith('/myapp/kv/'), handler);

// Ask the peerstore (filled by Identify) before opening:
final best = await host.peerStore.protoBook
    .firstSupportedProtocol(peer, ['/myapp/kv/2.0.0', '/myapp/kv/1.0.0']);
```

Bump the ID on any incompatible wire change. Add fields compatibly (protobuf)
within a version. `host.removeStreamHandler(id)` withdraws a protocol and
pushes the change to connected peers.

## Key concepts / API map

- `Host.setStreamHandler(ProtocolID, StreamHandler)`; `StreamHandler` is `Future<void> Function(P2PStream stream, PeerId remotePeer)`.
- `Host.setStreamHandlerMatch(pid, bool Function(ProtocolID) match, handler)`, `Host.removeStreamHandler(pid)`.
- `Host.newStream(PeerId, List<ProtocolID>, Context)`: dials if needed, waits for Identify, negotiates (bounded by the negotiation timeout, 10 s default).
- `P2PStream.read([maxLength])`: next chunk, at most `maxLength` bytes; empty at EOF.
- `P2PStream.write(Uint8List)`: waits for flow-control window; throws `StateError` after `closeWrite()`.
- `closeWrite()`: send FIN, keep reading. `closeRead()`: stop reading, pending read gets EOF.
- `close()`: FIN plus local teardown of both directions; no reads afterwards.
- `reset()`: abort both directions (Yamux RST); the peer's read ends with an error or EOF.
- `setDeadline(DateTime?)` (null clears), `setReadDeadline`, `setWriteDeadline`.
- `protocol()`, `id()`, `conn` (`remotePeer`, `remoteMultiaddr`), `stat()` (`direction`, `limited`), `scope()`, `isClosed`, `isWritable`.
- `stream.scope().setService('my.service')`, `reserveMemory(bytes, ReservationPriority.high)`: attribute resources to your service for the resource manager.
- `encodeVarint` / `decodeVarint` (`utils/varint.dart`).
- Exceptions: `YamuxStreamTimeoutException`, `YamuxStreamStateException` (`p2p/transport/multiplexing/yamux/yamux_exceptions.dart`), `ResourceLimitExceededException`.

## Gotchas

- **`close()` is not half-close.** After `close()` the stream is gone locally,
  and `read()` throws. A client that writes a request and calls `close()`
  never gets the reply. Use `closeWrite()`, read to EOF, then `close()`.
- **A server that reads to EOF needs the client's FIN.** If the handler
  loops until `read()` is empty, the client must `closeWrite()`, or both
  sides wait. go-libp2p and js-libp2p send FIN on a WINDOW_UPDATE frame;
  this is handled since 3.0.0.
- **One `read()` is not one message.** Chunks split and merge at any byte. Use a
  length prefix or read to EOF.
- **Always end every stream** with `close()` or `reset()`, on both sides and on
  every error path. Default limits allow 68 inbound streams per peer per
  protocol; leaked streams make the next ones fail with
  `ResourceLimitExceededException`. (Before 4.7.1 a stream closed by both
  sides but not read to EOF kept its scope; stay on a current version.)
- **Write deadlines are weak.** `setWriteDeadline` is stored but not enforced
  by Yamux streams, and `setDeadline` is checked only when a write starts. A
  write blocked on a peer that stopped reading waits. Wrap writes in
  `.timeout(...)` and `reset()` the stream on timeout.
- **Read deadlines:** with `setDeadline`/`setReadDeadline`, a read past the
  deadline throws `YamuxStreamTimeoutException` (or `YamuxStreamStateException`
  if the stream died meanwhile). Without one, a read gives up after 5 minutes.
  `MultiplexerConfig.streamReadTimeout` is not used by the stream code.
- **`Context(timeout: ...)` does not bound `newStream`.** Use `Future.timeout`
  around the call, or `Context().withDialPeerTimeout(d)` for the dial part only.
- **A handler that throws resets only its own stream** and is logged at
  WARNING (4.0.1). Still catch your own errors, and reset deliberately.
- **Handlers are not awaited by the muxer**: each runs concurrently. Guard
  shared state, and do not rely on handler order across streams.
- **Unsupported protocol**: `newStream` throws when the peer supports none of
  the IDs; check the peerstore first if you want a fallback without a round
  trip. Protocol IDs longer than about 1 KB fail multistream
  (`MessageTooLargeException`).
- **Bound every allocation by a size you chose.** A varint length comes from the
  peer; check it against `maxFrame` before buffering.
- **Relayed connections are limited.** A stream over `/p2p-circuit` shares the
  relay's per-connection budget (1 MiB and 1 h on a dart_libp2p relay, 128 KiB
  and 2 min on a go-libp2p relay). `stream.stat().limited` tells you. Keep bulk
  transfers for direct connections (see `dart-libp2p-nat-relay`).
- **New streams prefer a direct connection** to the peer over a relayed one
  (4.0.0). Do not cache a `Conn`; call `newStream` each time.
- **Prefer the documented `newStream(peer, pids, Context())`.** Older docs omit
  the `Context`, use `stream.stream`, or read `host.services`; none exist.

## Testing

Use two real hosts on loopback (see `dart-libp2p-host` Testing) and exercise
the protocol end to end:

```dart
test('kv round trip', () async {
  b.setStreamHandler('/myapp/kv/1.0.0', kvHandler);   // a, b: Hosts from setUp
  final reply = await request(a, b.id, Uint8List.fromList([1]))
      .timeout(const Duration(seconds: 10));
  expect(reply, isNotEmpty);
}, timeout: const Timeout(Duration(seconds: 30)));
```

- Test the edges: a peer that closes before replying (`next()` returns null),
  a frame over `maxFrame`, a handler that throws (the client should see EOF or
  an error, and the next stream must work), many sequential requests (catches
  leaked streams against the resource limits).
- Unit-test the framing without a network by feeding `FrameReader` a fake
  `P2PStream` whose `read()` returns scripted chunks, including splits inside
  the varint.
- For go-libp2p interop, use a TCP listen address; the repo's
  `interop/go-peer` shows a Go echo server (`/echo/1.0.0`) and the Dart side in
  `test/interop/go_interop_host_test.dart`.

## Red lines

- Never trust a length, count or ID from the peer without a limit.
- Never leave a stream open on an error path; `reset()` it.
- Never change the wire format under an existing protocol ID; add a new
  version and keep serving the old one while peers migrate.
- Never use reserved IDs (`/ipfs/...`, `/libp2p/...`, `/meshsub/...`) for your
  own protocol; namespace it (`/myapp/...`).
- Never block a handler on unbounded work while holding the stream; apply a
  deadline.

## Reference

- Protocols: https://cryptopeer.net/libp2p/core/protocols/
- Multiplexing (Yamux): https://cryptopeer.net/libp2p/core/multiplexing/
- Host API: https://cryptopeer.net/libp2p/core/host/
- Resource manager: https://cryptopeer.net/libp2p/core/resource-manager/
- Cookbook: https://cryptopeer.net/libp2p/core/cookbook/
- Source: https://github.com/stephanfeb/dart_libp2p (`lib/core/network/stream.dart`, `lib/p2p/transport/multiplexing/yamux/stream.dart`)