---
title: Connections and streams
description: 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:

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

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

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

| Connection target | Example |
| --- | --- |
| Peer ID; the endpoint applies its discovery-book and relay policy | `connect(peerId)` |
| One complete address ending in `/p2p/<peer-id>` | `connect("/ip4/127.0.0.1/udp/4001/quic-v1/p2p/12D3KooWDpJ7As7BWAwRMfu1VU2WCqNjvq387JEYKDBj4kx6nXTN")` |
| Several complete addresses naming one peer | `connect([quicAddr, tcpAddr])` |

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:

```ts
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.

```ts
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:

```ts
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.

```ts
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`:

```ts
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:

```bash
node receiver.mjs
```

In the same directory, create `sender.mjs`:

```ts
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();
}
```

```bash
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

```ts
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:

```ts
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:

**Node.js**

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

**React Native**

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

```ts
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:

```ts
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

| Operation | Result |
| --- | --- |
| `closeWrite()` | Graceful local half-close after earlier writes; reading remains possible. |
| `reset()` | Abruptly reset the stream and emit `closed`. |
| `abandon()` | Reset, relinquish the handle, and suppress later events. |
| `stream[Symbol.dispose]()` | Equivalent to `abandon()`. |

`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](/docs/typescript/networking-and-events).