# Web Streams

NAPI-RS can accept a JavaScript ReadableStream or WritableStream as a function argument, and can create a ReadableStream from any Rust Stream and hand it back to JavaScript.

Support lives behind the web_stream Cargo feature, which also enables tokio_rt (see Cargo features):

[dependencies]
napi = { version = "3", features = ["async", "web_stream"] }
napi-derive = "3"

The host runtime must provide compatible Web Streams globals — Node.js 18+ exposes ReadableStream globally (also via node:stream/web). When the global ReadableStream constructor is missing, stream creation fails with a "ReadableStream is not supported in this Node.js version" error.

# Accepting a ReadableStream

A ReadableStream<T> argument is validated with instanceof against the global ReadableStream constructor. Calling .read() locks the stream with getReader() and returns a Reader<T> — a Rust Stream of Result<T> items you can consume with tokio_stream::StreamExt (re-exported as napi::tokio_stream) inside any async block:

use napi::bindgen_prelude::*;
use napi_derive::napi;
use napi::tokio_stream::StreamExt;

/// Counts the chunks read from a stream, swallowing (dropping) any read error.
#[napi]
pub fn drain_stream_count(
  env: &Env,
  stream: ReadableStream<Uint8Array>,
) -> Result<AsyncBlock<u32>> {
  let mut reader = stream.read()?;
  AsyncBlockBuilder::new(async move {
    let mut count = 0u32;
    while let Some(item) = reader.next().await {
      match item {
        Ok(_) => count += 1,
        // Drop the error on the Tokio thread instead of returning it to JS.
        Err(_err) => break,
      }
    }
    Ok(count)
  })
  .build(env)
}

export declare function drainStreamCount(
  stream: ReadableStream<Uint8Array>,
): Promise<number>

import { createReadStream } from 'node:fs'
import { Readable } from 'node:stream'
import { drainStreamCount } from './index.js'

// Any Node.js stream can be converted with Readable.toWeb
const count = await drainStreamCount(
  Readable.toWeb(createReadStream('./file.txt')),
)

# Returning a ReadableStream

To hand a stream to JavaScript, build one from a Rust Stream<Item = Result<T>> that is Unpin + Send + 'static:

• ReadableStream::new(env, stream) — for any item type that implements ToNapiValue + Send + 'static (numbers, strings, #[napi(object)] structs, …).
• ReadableStream::create_with_stream_bytes(env, stream) — for byte streams; the items only need Into<Vec<u8>>, and the underlying source is created with type: 'bytes'.
• ReadableStream::with_stream_bytes_and_readable_stream_class(env, class, stream) — the same byte stream, but constructed from a caller-supplied ReadableStream class, for runtimes whose stream implementation is not the global constructor.

use napi::bindgen_prelude::*;
use napi_derive::napi;
use napi::tokio_stream::wrappers::ReceiverStream;

#[napi]
pub fn create_readable_stream(env: &Env) -> Result<ReadableStream<'_, BufferSlice<'_>>> {
  let (tx, rx) = napi::tokio::sync::mpsc::channel(100);
  std::thread::spawn(move || {
    for _ in 0..100 {
      tx.try_send(Ok(b"hello".to_vec())).expect("stream queue is full or closed");
    }
  });
  ReadableStream::create_with_stream_bytes(env, ReceiverStream::new(rx))
}

#[napi(object)]
#[derive(Default)]
pub struct StreamItem {
  pub name: String,
  pub size: i32,
}

/// Creates a ReadableStream that emits StreamItem objects.
#[napi]
pub fn create_readable_stream_with_object(
  env: &Env,
) -> Result<ReadableStream<'_, StreamItem>> {
  let (tx, rx) = napi::tokio::sync::mpsc::channel(100);
  std::thread::spawn(move || {
    for i in 0..100 {
      let item = StreamItem {
        name: format!("item-{i}"),
        size: i,
      };
      tx.try_send(Ok(item)).expect("stream queue is full or closed");
    }
  });
  ReadableStream::new(env, ReceiverStream::new(rx))
}

export declare function createReadableStream(): ReadableStream<Buffer>

export declare function createReadableStreamWithObject(): ReadableStream<StreamItem>

On the JavaScript side a returned stream is a real Web ReadableStream — consume it with a reader or for await:

import { createReadableStreamWithObject } from './index.js'

for await (const item of createReadableStreamWithObject()) {
  console.log(item.name, item.size)
}

ReadableStream also exposes the locked() and cancel(reason) Web API methods from Rust.

# AsyncBlock: a promise with a dispose hook

The examples above return AsyncBlock<T>, the primitive behind "return a promise" when you are not using an async fn. An AsyncBlock<T> starts its future on the NAPI-RS runtime eagerly — when the Rust function is called, not when JavaScript awaits it — and the returned value converts into a JavaScript Promise<T> (see Promise and async fn).

Build one with AsyncBlockBuilder:

• AsyncBlockBuilder::new(future) — the future is Future<Output = Result<V>> + Send + 'static.
• .with_dispose(|env| ...) — a hook that runs on the JavaScript thread when the future resolves successfully, before the value is converted. Note it only covers the Ok path: when the future returns Err, the promise rejects without running the hook. If a resource must be released on failure too, release it inside the future itself (or map the Err before building the block) and keep with_dispose for the success path.
• .build(env) — spawns the future and produces the AsyncBlock<V>.
• AsyncBlockBuilder::build_with_map(env, future, map) — a static constructor whose map closure runs on the JavaScript thread at resolution time, turning the future's output into the final JavaScript value. This is the escape hatch for values that can only be created on the JS thread.

build_with_map is how you resolve a promise with a zero-copy buffer: the future produces owned bytes on the runtime thread, and the map closure wraps them in a BufferSlice<'static> — without copying — on the JavaScript thread:

use napi::bindgen_prelude::*;
use napi_derive::napi;
use napi::tokio_stream::StreamExt;

#[napi]
pub fn accept_stream(
  env: &Env,
  stream: ReadableStream<Uint8Array>,
) -> Result<AsyncBlock<BufferSlice<'static>>> {
  let mut reader = stream.read()?;
  AsyncBlockBuilder::build_with_map(
    env,
    async move {
      let mut bytes = Vec::new();
      while let Some(chunk) = reader.next().await {
        bytes.extend_from_slice(&chunk?);
      }
      Ok(bytes)
    },
    |env, mut value| {
      let value_ptr = value.as_mut_ptr();
      unsafe {
        BufferSlice::from_external(&env, value_ptr, value.len(), value, move |_, bytes| {
          drop(bytes);
        })
      }
    },
  )
}

(The full example adapts the chunk stream into an AsyncRead with tokio_util::io::StreamReader instead of collecting chunk by chunk; both shapes end in the same from_external map closure.)

export declare function acceptStream(
  stream: ReadableStream<Uint8Array>,
): Promise<Buffer>

Because the map closure runs on the JavaScript thread with a live Env, it can also call back into JavaScript constructors. The full fetch example uses this to build a Web Response around a NAPI-RS byte stream, and overrides the generated declaration with ts_return_type so the return type references the undici-types package:

#[napi(ts_return_type = "Promise<import('undici-types').Response>")]
pub fn fetch(
  env: &Env,
  url: String,
  request_init: Option<RequestInit>,
) -> Result<AsyncBlock<Unknown<'static>>> {
  AsyncBlockBuilder::build_with_map(
    env,
    async move { /* ... reqwest request ... */ },
    |env, response| {
      let global = env.get_global()?;
      let response_constructor: Function<ReadableStream<BufferSlice>, ()> =
        global.get_named_property("Response")?;
      let js_stream = ReadableStream::create_with_stream_bytes(&env, /* response body stream */)?;
      response_constructor.new_instance(js_stream)
    },
  )
}

export declare function fetch(
  url: string,
  requestInit?: RequestInit | undefined | null,
): Promise<import('undici-types').Response>

# Accepting a WritableStream

A WriteableStream argument (note the NAPI-RS spelling) wraps a JavaScript WritableStream. It is a thin handle over the Web API: ready(), abort(reason), and close() each call the corresponding JavaScript method and return a PromiseRaw<()> you can convert into an awaitable Promise inside an async context. The writable side of a stream cannot be created from Rust today — accept it from JavaScript, or pipe a returned ReadableStream into it on the JavaScript side.