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, wait for Identify readiness, then open a negotiated stream.

Connect with a Promise

Use connectAddr when you have one complete peer multiaddress:

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

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

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

All three resolve with { connectId, peerId, path }. The first path may be directDialed, directPunched, or relayed. Call path(peerId) when you need the current answer; a relayed connection can upgrade after the original Promise resolves.

The lower-level dial, dialIp4, and dialIp6 methods only start direct TCP or QUIC dials and return connection IDs immediately. They do not run the relay race or return a Path. Observe peerReady or call waitPeerReady afterward.

Timeouts and cancellation

connect*, 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.

const controller = new AbortController();

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

// A user action or a shutdown path can stop this attempt.
controller.abort();
await connectPromise; // rejects with AbortError

For a connect* Promise, an abort or timeout cancels the underlying connection attempt. 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 connection-attempt ID:

const connectId = endpoint.startConnectAddr(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);

One terminal result can be consumed once. Waiting twice, or waiting for an unknown attempt, rejects with ConnectResultUnavailableError.

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. Wait for peerReady before opening so the remote Identify snapshot is available.

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(),
  transports: {
    quic: { 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}`);
  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.connectAddr(remoteAddress, {
    timeoutMs: 10_000,
  });
  await endpoint.waitPeerReady(connected.peerId, { timeoutMs: 10_000 });

  const stream = await endpoint.openStream(connected.peerId, protocol, {
    timeoutMs: 10_000,
  });
  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 }
);

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

write accepts UTF-8 text, Uint8Array, or ArrayBuffer. closeWrite() half-closes only the local write side; reads stay available until the remote half-closes or the stream terminates.

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. Pull reads use a bounded buffer; if the application does not keep up, dataOverflow reports dropped chunks and bytes. Treat that as a protocol-level recovery condition rather than silently continuing a framed transfer.

Next: Networking and events.