Streaming

Explains how to store and read large data with streams.

Choosing types

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

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

You can pass either a plain value or a ReadableStream.

Save a single value

Pass a plain value for small data.

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

Save with a stream

Pass a ReadableStream to write large data incrementally.

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

Reading

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

Use for fine-grained control over each chunk.

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

Use for concise code.

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

Use for automatic release of the read lock.

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

Notes by Storage

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

Long transfers can be interrupted with an AbortSignal.

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

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