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

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

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