water - Alternative to ReadableStream, WritableStream and TransformStream.

Documentation Index

This library introduces 3 classes: RdStream, WrStream and TrStream, that can be used in place of ReadableStream<Uint8Array>, WritableStream<Uint8Array> and TransformStream<Uint8Array, Uint8Array>.

This library reimplements ReadableStream, WritableStream and TransformStream in the fashion that the author of this library likes. The style of this library is to reuse buffers, and not to transfer buffers.

This library requires to implement only a simple interface to create stream of bytes. Here is such interface that any object can implement to provide to others a readable byte-stream:

interface Source
{	// Reads up to `p.byteLength` bytes into `p`. It resolves to the number of
    // bytes read (`0` < `n` <= `p.byteLength`) and rejects if any error
    // encountered. Even if `read()` resolves to `n` < `p.byteLength`, it may
    // use all of `p` as scratch space during the call. If some data is
    // available but not `p.byteLength` bytes, `read()` conventionally resolves
    // to what is available instead of waiting for more.
    //
    // When `read()` encounters end-of-file condition, it resolves to EOF
    // (`null`).
    //
    // When `read()` encounters an error, it rejects with an error.
    //
    // Callers should always process the `n` > `0` bytes returned before
    // considering the EOF (`null`). Doing so correctly handles I/O errors that
    // happen after reading some bytes and also both of the allowed EOF
    // behaviors.
    //
    // Implementations should not retain a reference to `p`.
    read(p: Uint8Array): Promise<number|null>;
}

Implementing this interface is enough, however this is only a subset of the full Source interface that is shown below.

Example of how you can implement it:

const rdStream = new RdStream
(	{	async read(p)
        {	// ...
            // Load data to `p`
            // ...
            return p.byteLength; // or less
        }
    }
);

// now use `rdStream` as you would use an instance of ReadableStream<Uint8Array>

And the following interface is for writeable streams:

interface Sink
{	// Writes `p.byteLength` bytes from `p` to the underlying data stream. It
    // resolves to the number of bytes written from `p` (`0` <= `n` <=
    // `p.byteLength`) or reject with the error encountered that caused the
    // write to stop early. `write()` must reject with a non-null error if
    // would resolve to `n` < `p.byteLength`. `write()` must not modify the
    // slice data, even temporarily.
    //
    // Implementations should not retain a reference to `p`.
    write(p: Uint8Array): Promise<number>;
}

Example:

const wrStream = new WrStream
(	{	async write(p)
        {	// ...
            // Write `p` somewhere
            // ...
            return p.byteLength; // or less
        }
    }
);

// now use `wrStream` as you would use an instance of WritableStream<Uint8Array>

Differences from ReadableStream, WritableStream and TransformStream

No controllers concept.
BYOB-agnostic. Data consumer can use BYOB or regular reading mode, and there's no need of handling these situations differently.
No transferring buffers that you pass to reader.read(buffer), so the buffers remain usable after the call.

Additional features:

RdStream.cancel() and WrStream.abort() work also on locked streams.
getReader() and getWriter() have getReaderWhenReady() and getWriterWhenReady() counterparts, that wait for reader/writer to be unlocked.
RdStream.values(), RdStream.tee(), RdStream.pipeTo() and RdStream.pipeThrough() are present in both RdStream and Reader.
Also RdStream and Reader have additional methods: uint8Array(), text() and unread().
RdStream.pipeTo() and RdStream.pipeThrough() are restartable (Transformer.transform() can close it's writer, and then the rest of the input stream can be piped to elsewhere).
Reader and Writer implement Symbol.dispose that releases the lock.
WrStream has flush feature.

Exported classes and types

import {RdStream, Source} from 'https://deno.land/x/water@v1.0.23/mod.ts';
import {WrStream, Sink} from 'https://deno.land/x/water@v1.0.23/mod.ts';
import {TrStream, Transformer} from 'https://deno.land/x/water@v1.0.23/mod.ts';
import {TooBigError} from 'https://deno.land/x/water@v1.0.23/mod.ts';

class RdStream

This class extends ReadableStream<Uint8Array>.

An instance can be created from Source definition (object that has read() method):

const rdStream = new RdStream
(	{	async read(p)
        {	// ...
            // Load data to `p`
            // ...
            return p.byteLength; // or less
        }
    }
);

For example Deno.stdin implements read():

const rdStream = new RdStream(Deno.stdin);

Or it can be created as wrapper on existing ReadableStream object. Here is another way of creating RdStream that reads from stdin:

const rdStream = RdStream.from(Deno.stdin.readable);

Now rdStream and Deno.stdin.readable are the same by the means of ReadableStream (both have getReader()), but RdStream also has features that ReadableStream doesn't have. For example text() function:

console.log(await rdStream.text());

Creating RdStream from read() implementors (like new RdStream(Deno.stdin)) is preferrable (because it works faster) than creating from another streams (like RdStream.from(Deno.stdin.readable)). However note that Deno.stdin also implements close(), so the file descriptor will be closed after reading to the end. To prevent this, use:

const rdStream = new RdStream({read: p => Deno.stdin.read(p)});

RdStream.from() also allows to create RdStream instances from iterable objects that yield Uint8Array items (see RdStream.from()).

Example

The following example demonstrates readable stream that streams the string provided to it's constructor.

import {RdStream} from 'https://deno.land/x/water@v1.0.23/mod.ts';

const textEncoder = new TextEncoder;

// Readable stream, compatible with `ReadableStream<Uint8Array>`, that streams the string provided to it's constructor.
class StringStreamer extends RdStream
{	constructor(str='')
    {	let data = textEncoder.encode(str);
        super
        (	{	read(view)
                {	const n = Math.min(view.byteLength, data.byteLength);
                    view.set(data.subarray(0, n));
                    data = data.subarray(n);
                    return n;
                }
            }
        );
    }
}

// Write word "Hello" to stdout
await new StringStreamer('Hello\n').pipeTo(Deno.stdout.writable);

Constructor:

🔧 RdStream.constructor(source: Source)

type Source =
{
📄 autoAllocateChunkSize?: number
📄 autoAllocateMin?: number
⚙ start?(): void | PromiseLike<void>
⚙ read(view: Uint8Array): number | PromiseLike<number>
⚙ close?(): void | PromiseLike<void>
⚙ cancel?(reason: unknown): void | PromiseLike<void>
⚙ catch?(reason: unknown): void | PromiseLike<void>
⚙ finally?(): void | PromiseLike<void>
}

RdStream instances are constructed from Source objects, that have definition of how data stream is generated.

If there's start() method, it gets called immediately, even before the constructor returns, to let the stream generator to initialize itself. If start() returned Promise, this Promise is awaited before calling read() for the first time.

The only mandatory Source method is read(). This method is called each time data is requested from the stream by consumer. Calls to read() are sequential, and new call doesn't begin untill previous call is finished (it's promise is fulfilled). When read() is called it must load bytes to provided buffer, and return number of bytes loaded. To indicate EOF it can return either 0 or null. It can return result asynchronously (Promise object) or synchronously (number or null result).

Stream consumer can read the stream in regular or "BYOB" mode. In BYOB, the consumer provides it's own buffer, which is passed to Reader.read(). This buffer can be of any non-zero size. In regular mode a buffer of at least autoAllocateMin bytes is allocated (and passed to Reader.read()). The maximum auto-allocated buffer size is autoAllocateChunkSize.

When read() returned EOF (0 or null), close() gets called to finalize the stream generator.

If read() thrown exception, catch() is called instead of close(). Also if read() successfully returned EOF, but then close() thrown exception, catch() is also called.

Stream consumer can decide to cancel the stream by calling RdStream.cancel() or Reader.cancel(). In this case cancel() callback gets called. This is the only callback that can be called in the middle of read() work, when asynchronous read() didn't return, so it can tell read() to return earlier. If cancel() thrown exception, catch() is called as the last action.

Properties:

📄 get RdStream.locked(): boolean

When somebody wants to start reading this stream, he calls RdStream.getReader(), and after that call the stream becomes locked. Future calls to RdStream.rdStream.getReader() will throw error till the reader is released (Reader.releaseLock()).

Other operations that read the stream (like RdStream.pipeTo()) also lock it (internally they get reader, and release it later).

📄 get RdStream.isClosed(): boolean

Becomes true when the stream is read to the end, or after RdStream.cancel() was called.

Methods:

⚙ RdStream.getReader(_options?: {mode?: "byob"}): (ReadableStreamBYOBReader | ReadableStreamDefaultReader<Uint8Array>) & Omit<Reader, "read">

Returns object that allows to read data from the stream. The stream becomes locked till this reader is released by calling Reader.releaseLock() or Symbol.dispose().

If the stream is already locked, this method throws error.

⚙ RdStream.getReaderWhenReady(_options?: {mode?: "byob"}): Promise<(ReadableStreamBYOBReader | ReadableStreamDefaultReader<Uint8Array>) & Omit<Reader, "read">>

Like getReader(), but waits for the stream to become unlocked before returning the reader (and so locking it again).

⚙ RdStream.cancel(reason?: unknown): Promise<void>

Interrupt current reading operation (reject the promise that Reader.read() returned, if any), and tell to discard further data in the stream. This leads to calling Source.cancel(reason), even if current Source.read() didn't finish. Source.cancel() must implement the actual behavior on how to discard further data, and finalize the source, as no more callbacks will be called.

In contrast to ReadableStream.cancel(), this method works even if the stream is locked.

⚙ RdStream.unread(chunk: Uint8Array): void

Push chunk to the stream, so next read will get it. This creates internal buffer, and copies the chunk contents to it.

⚙ RdStream.[Symbol.asyncIterator](options?: {preventCancel?: boolean}): ReadableStreamIterator

⚙ RdStream.values(options?: {preventCancel?: boolean}): ReadableStreamIterator

Allows to iterate this stream yielding Uint8Array data chunks.

Usually you want to use for await...of to iterate.

for await (const chunk of rdStream)
{	// ...
}

It's also possible to iterate manually. In this case you need to be "using" the iterator, or to call releaseLock() explicitly.

using it = rdStream.values();
while (true)
{	const {value, done} = await it.next();
    if (done)
    {	break;
    }
    // ...
}