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

Use `connectAddr` when you have one complete peer multiaddress:

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

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

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

| Input you have | Method |
| --- | --- |
| Peer ID already in the discovery book | `connect(peerId, options?)` |
| Peer ID plus an ordered candidate list | `connectWithAddrs(peerId, addresses, options?)` |
| One address ending in `/p2p/<peer-id>` | `connectAddr(address, options?)` |

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:

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

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

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

```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(),
  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:

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

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

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:

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