---
title: "Streaming"
description: "Explains how to store and read large data with streams."
---

## Choosing types [#select-types]

Keys used with streams are defined with `StreamValue<T>` or `Value<T>`. Keys defined with `PlainValue<T>` cannot use `stream()`.

```ts
const kvs = UniKvs.config<{
  logs: StreamValue<Uint8Array>;
  blob: Value<Uint8Array>;
}>()
  .appendStorage(new Memory())
  .create();
```

| Type | `set` | `get` | `stream` |
| --- | --- | --- | --- |
| `PlainValue<T>` | Supported. | Supported. | Not supported. |
| `StreamValue<T>` | Supports values or streams. | Not supported. | Supported. |
| `Value<T>` | Supports values or streams. | Supported. | Supported. |

When `schema` is specified, each chunk is validated. Bad write chunks are reported as `PluginOperationAggregateError` containing `InvalidInputError`, and bad read chunks are reported as `InvalidOutputError`.

## Writing [#write]

You can pass either a plain value or a `ReadableStream`.

1. **Save a single value**

    Pass a plain value for small data.

    ```ts
    await kvs.set("logs", new Uint8Array([1, 2, 3]));
    ```

2. **Save with a stream**

    Pass a `ReadableStream` to write large data incrementally.

    ```ts
    const src = new ReadableStream({
      start(controller) {
        controller.enqueue(new Uint8Array([1]));
        controller.enqueue(new Uint8Array([2]));
        controller.close();
      },
    });
    await kvs.set("logs", src);
    ```

## Reading [#read]

`stream()` returns a `ValueStream`. Choose from three patterns for your use case.

**getReader**

Use for fine-grained control over each chunk.

```ts
const reader = (await kvs.stream("logs")).getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    break;
  }
  console.log(value);
}
```

**for-await**

Use for concise code.

```ts
for await (const chunk of await kvs.stream("logs")) {
  console.log(chunk);
}
```

**await-using**

Use for automatic release of the read lock.

```ts
await using (const valueStream = await kvs.stream("logs")) {
  const reader = valueStream.getReader();
  const { value } = await reader.read();
  console.log(value);
}
```

:::warning
Dispose of streams when you are done with them. `await using` releases the lock automatically.
:::

## Notes by Storage [#storages]

:::warning
Memory streams are implemented by concatenating chunks. They are not suitable for sequential processing of very large data.
:::

- Files, S3, and OPFS are suitable for streaming byte data. See each package reference for details.
- IndexedDB can store arbitrary data, but check the implementation constraints for stream handling.
- Some Transformers do not support stream conversion. See each package reference for support status.

## Cancellation [#cancellation]

Long transfers can be interrupted with an `AbortSignal`.

```ts
const ac = new AbortController();
setTimeout(() => ac.abort(), 1000);

await kvs.set("logs", src, { signal: ac.signal });
```