Connections and streams

Connect with cancellable Promises, inspect direct or relayed paths, and exchange bytes over custom protocol streams.

The TypeScript SDK separates finding a usable network path from opening an application protocol. Connect first. Call waitPeerReady when you need Identify’s advertised protocols; openStream is allowed once the path is usable.

Connect with a Promise

Pass connect a Connection target. The simplest is one complete peer multiaddress, copied as advertised:

const result = await endpoint.connect(remoteAddress, {
  timeoutMs: 15_000,
});

await endpoint.waitPeerReady(result.peerId, {
  timeoutMs: 10_000,
});

console.log(result.connectId, result.path);

A string that starts with / is an address; any other string is a peer ID. Transport and IP family come from the addresses, and every candidate races under one attempt. Malformed targets throw synchronously; a valid peer ID with no route is admitted and then rejects with ConnectFailedError.

connect resolves with { connectId, peerId, path }. The first path may be direct, punched, or relayed. Call path(peerId) when you need the current answer; a relayed connection can upgrade after the original Promise resolves. An upgrade arrives as connectionReplaced ({ peerId, oldConnId, newConnId }): the peer stays connected, Identify runs again on newConnId, and waitPeerReady waits for the new connection’s peerReady. Streams on oldConnId end when the endpoint processes the replacement: pending and later reads reject with StreamClosedError. Until then read() can still return buffered data or a pending Promise, while write rejects and closeWrite and reset throw rather than reaching a stream on newConnId. A closeWrite queued behind pending writes cannot throw; if native then refuses it, the stream’s terminal event reports the end.

Timeouts and cancellation

connect, waitConnectResult, waitPeerReady, ping, openStream, once, and waitFor accept:

interface OpOptions {
  timeoutMs?: number;
  signal?: AbortSignal;
}

The default timeout is 65 seconds. Set timeoutMs: 0 only when a caller truly wants an unbounded wait.

For a Connection attempt the two controls differ on purpose:

• Timeout bounds only the local wait. It rejects with TimeoutError and the attempt keeps running. Pass cancelOnTimeout: true to also cancel it.
• Abort cancels the attempt. The wait then settles from the attempt’s terminal: ConnectCancelledError (an AbortError), or the real result if the attempt had already connected.

const controller = new AbortController();

const connectPromise = endpoint.connect(remoteAddress, {
  signal: controller.signal,
  timeoutMs: 20_000,
});

// A user action or a shutdown path can stop this attempt.
controller.abort();
// ConnectCancelledError only if the attempt has not already connected when
// abort takes effect. A connection that completes first still resolves.
await connectPromise;

Waiting never blocks event dispatch: on handlers, events() iterators, and other waits keep receiving unrelated events. Concurrent ping calls to the same peer share one Ping request while each caller keeps its own timeout and abort signal.

Use the split-phase API when the caller must retain the Connect ID:

const connectId = endpoint.startConnect(remoteAddress);
const resultPromise = endpoint.waitConnectResult(connectId, {
  timeoutMs: 20_000,
});

// A separate action can cancel by ID.
const cancel = () => endpoint.cancelConnect(connectId);

const result = await resultPromise;
console.log(result.path);

Every attempt ends with exactly one terminal event: pathEstablished, connectFailed, or connectCancelled. cancelConnect does not settle a wait by itself; the terminal does. A pending wait rejects with ConnectCancelledError (an AbortError), or resolves normally if the attempt had already connected. Cancelling never closes an established connection; use disconnect(peerId) for that.

One terminal result can be consumed once, and one wait per attempt may be pending. Waiting twice, or waiting for an unknown attempt, rejects with ConnectResultUnavailableError. After a timeout, call waitConnectResult again to pick up the outcome.

Event delivery loss

Event queues are bounded. If a terminal event is dropped, the affected wait rejects with ConnectResultLostError instead of hanging. State getters stay authoritative: call connectedPeers(), path(peerId), or connectionInfo(peerId) to recover the current state.

Register a protocol

Protocol IDs are versioned strings owned by your application. Register them in the initial configuration or before a remote peer opens an inbound stream.

const endpoint = Minip2p.create({
  secretKey,
  protocols: ["/example/files/1"],
});

endpoint.addProtocol("/example/chat/1");

Both peers must support the same ID. waitPeerReady is useful when you want the remote Identify snapshot before opening; it is not required to start negotiation on a known protocol.

Exchange one request and response

This Node.js example uses one stream for a request and its reply. From a directory where you installed @minip2p/node, create receiver.mjs:

import { Minip2p, generateSecretKey } from "@minip2p/node";

const protocol = "/example/echo/1";
const decoder = new TextDecoder();
const endpoint = Minip2p.create({
  protocols: [protocol],
  secretKey: generateSecretKey(),
  listen: ["/ip4/127.0.0.1/udp/4002/quic-v1"],
});

try {
  console.log(endpoint.listenAddrs()[0]);

  const stream = await endpoint.once("stream", { timeoutMs: 0 });
  let request = "";
  for await (const chunk of stream) {
    request += decoder.decode(chunk, { stream: true });
  }
  request += decoder.decode();

  console.log(`received: ${request}`);
  await stream.write(`echo: ${request}`);
  stream.closeWrite();

  await new Promise((resolve) => stream.on("closed", resolve));
} finally {
  endpoint.close();
}

Run the receiver in the first terminal and copy its complete printed address:

node receiver.mjs

In the same directory, create sender.mjs:

import { Minip2p, generateSecretKey } from "@minip2p/node";

const remoteAddress = process.argv[2];
if (remoteAddress === undefined) {
  throw new Error("pass the receiver's peer multiaddress");
}

const protocol = "/example/echo/1";
const decoder = new TextDecoder();
const endpoint = Minip2p.create({
  protocols: [protocol],
  secretKey: generateSecretKey(),
});

try {
  const connected = await endpoint.connect(remoteAddress, {
    timeoutMs: 10_000,
  });
  await endpoint.waitPeerReady(connected.peerId, { timeoutMs: 10_000 });

  const stream = await endpoint.openStream(connected.peerId, protocol, {
    timeoutMs: 10_000,
  });
  await stream.write("hello");
  stream.closeWrite();

  let reply = "";
  for await (const chunk of stream) {
    reply += decoder.decode(chunk, { stream: true });
  }
  console.log(reply + decoder.decode());
} finally {
  endpoint.close();
}

node sender.mjs /ip4/127.0.0.1/udp/4002/quic-v1/p2p/12D3KooW...

Run the sender in the second terminal with the receiver’s address. The receiver prints received: hello. The sender prints echo: hello.

The sender closes its endpoint after reading the complete reply. The receiver waits for the stream to close before stopping its own endpoint.

Open and write

const stream = await endpoint.openStream(peerId, "/example/files/1", {
  timeoutMs: 10_000,
});

await stream.write("header\n");
await stream.write(new Uint8Array([1, 2, 3]));
stream.closeWrite();

write accepts UTF-8 text, Uint8Array, or ArrayBuffer and returns a Promise that resolves once the native endpoint has accepted every byte. A peer that reads slowly makes writes resolve later; nothing is dropped and nothing tears down. Writes go out in call order. Each stream buffers up to stream.writeHighWaterMark bytes (256 KiB by default, counting the write still in flight); a write past that rejects with WriteBufferFullError, so await earlier writes before queueing more. A write into an empty buffer is always admitted, whatever its size.

closeWrite() half-closes only the local write side, after every earlier write has been accepted, so write(); closeWrite() delivers every byte before the FIN. Later writes reject. Reads stay available until the remote half-closes or the stream terminates. If the remote stops reading the stream, pending writes reject and the stream emits writeStopped.

Claim inbound streams

A named stream handler receives the live handle:

const removeInbound = endpoint.on("stream", (stream) => {
  if (stream.protocolId !== "/example/files/1") {
    stream.abandon();
    return;
  }

  void receiveFile(stream);
});

Unclaimed inbound streams are abandoned. A catch-all event subscriber receives only inboundStream metadata and cannot claim the live handle.

Choose one read mode

Pull mode is natural for framed or sequential decoders. The Stream type comes from the binding package for your runtime:

import type { Stream } from "@minip2p/node";

import type { Stream } from "@minip2p/react-native";

async function receiveFile(stream: Stream) {
  try {
    for (;;) {
      const chunk = await stream.read();
      if (chunk === undefined) break;
      consume(chunk);
    }
    stream.closeWrite();
  } catch (error) {
    stream.reset();
    throw error;
  }
}

Flowing mode delivers chunks to a handler:

const removeData = stream.on("data", (chunk) => consume(chunk));
const removeRemoteClose = stream.on("remoteWriteClosed", () => {
  removeData();
  finishIncomingMessage();
});

// Call this if the owner goes away first; both callbacks are idempotent.
const stopFlowing = () => {
  removeData();
  removeRemoteClose();
};

Pull reads and flowing data handlers are mutually exclusive. Calling read() after subscribing to data, or subscribing to data after the first read(), rejects or throws.

End a stream deliberately

remoteWriteClosed means no more inbound bytes will arrive. closed is the terminal lifecycle event.

Reading applies backpressure

Received bytes are never dropped. A chunk counts as consumed when read() returns it, or when every data handler has returned from it, and each stream delivers at most one receive window of bytes that have not been consumed. A reader that stops reading therefore stalls its sender instead of buffering without bound, and the sender resumes as the reader catches up.

An async data handler is consumed when it returns its Promise, not when that Promise settles. To hold the sender back while asynchronous work on a chunk runs, use read() (or for await) and await the work before reading the next chunk.

An unread stream keeps its stream slot after it closes, until its bytes are read or the stream is reset or abandoned, so a peer cannot grow buffered data by opening and closing streams.

Next: Networking and events.