water

Alternative to ReadableStream, WritableStream and TransformStream.

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:

reader.cancel() and writer.abort() work also on locked streams.
getReader() and getWriter() have getReaderWhenReady() and getWriterWhenReady() counterparts, that wait for reader/writer to be unlocked.
values(), tee(), pipeTo() and pipeThrough() are present in both RdStream and Reader.
Also RdStream and Reader have additional methods: uint8Array(), text() and unread().
pipeTo() and pipeThrough() are restartable (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.13/mod.ts';
import {WrStream, Sink} from 'https://deno.land/x/water@v1.0.13/mod.ts';
import {TrStream, Transformer} from 'https://deno.land/x/water@v1.0.13/mod.ts';
import {TooBigError} from 'https://deno.land/x/water@v1.0.13/mod.ts';

RdStream
WrStream
TrStream

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.13/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:

function RdStream.constructor(source: Source);

type Source =
{	/**	When auto-allocating (reading in non-byob mode) will pass to {@link Source.read} buffers of at most this size.
        If undefined or non-positive number, a predefined default value (like 32 KiB) is used.
     **/
    autoAllocateChunkSize?: number;

    /**	When auto-allocating (reading in non-byob mode) will not call `read()` with buffers smaller than this.
        First i'll allocate `autoAllocateChunkSize` bytes, and if `read()` callback fills in only a small part of them
        (so there're >= `autoAllocateMin` unused bytes in the buffer), i'll reuse that part of the buffer in next `read()` calls.
     **/
    autoAllocateMin?: number;

    /**	This callback is called immediately during `RdStream` object creation.
        When it's promise resolves, i start to call `read()` to pull data as response to `reader.read()`.
        Only one call is active at each moment, and next calls wait for previous calls to complete.

        At the end one of `close()`, `cancel(reason)` or `catch(error)` is called.
        - `close()` is called if `read()` returned EOF (`0` or `null`).
        - `cancel()` if caller called `rdStream.cancel(reason)` or `reader.cancel(reason)`.
        - `catch()` if `read()` thrown exception or returned a rejected promise.

        And the very last step is to call `finally()`, and if it thrown also to call `catch()` (again?).
     **/
    start?(): void | PromiseLike<void>;

    /**	This method is called to pull data from input source to a Uint8Array object provied to it.
        The object provided is never empty.
        The function is expected to load available data to the view, and to return number of bytes loaded.
        On EOF it's expected to return `0` or `null`.
        This callback is called as response to user request for data, and it's never called before such request.
     **/
    read(view: Uint8Array): number | null | PromiseLike<number|null>;

    /**	This method is called when {@link Source.read} returns `0` or `null` that indicate EOF.
        After that, no more callbacks are called (except `catch()`).
        If you use `Deno.Reader & Deno.Closer` as source, that source will be closed when read to the end without error.
     **/
    close?(): void | PromiseLike<void>;

    /**	Is called as response to `rdStream.cancel()` or `reader.cancel()`.
        After that, no more callbacks are called (except `catch()`).
        If this callback is not set, the default behavior is to read and discard the stream to the end.
     **/
    cancel?(reason: any): void | PromiseLike<void>;

    /**	Is called when `start()`, `read()`, `close()` or `cancel()` thrown exception or returned a rejected promise.
        After that, no more callbacks are called.
        Exceptions in `catch()` are silently ignored.
     **/
    catch?(reason: any): void | PromiseLike<void>;

    /**	Is called when the stream is finished in either way.
     **/
    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 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 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:

locked

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

isClosed

readonly RdStream.isClosed: boolean;

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

Methods:

getReader

function RdStream.getReader(options?: {mode?: undefined}): ReadableStreamDefaultReader<Uint8Array> & Reader;
function RdStream.getReader(options: {mode: 'byob'}): ReadableStreamBYOBReader & Reader;

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

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

getReaderWhenReady

function RdStream.getReaderWhenReady(options?: {mode?: undefined}): Promise<ReadableStreamDefaultReader<Uint8Array> & Reader>;
function RdStream.getReaderWhenReady(options: {mode: 'byob'}): Promise<ReadableStreamBYOBReader & Reader>;

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

cancel

function RdStream.cancel(reason?: any): 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.

unread

function 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.

[Symbol.asyncIterator], values

function RdStream[Symbol.asyncIterator](options?: {preventCancel?: boolean});
function RdStream.values(options?: {preventCancel?: boolean});

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;
    }
    // ...
}