Use Web Streams API for AdbScoket

[yume-chan]

1. Current design

Now AdbSocket is a ordinary class.

export declare class AdbSocket implements AdbSocketInfo {
    private readonly controller;

    get backend(): AdbBackend;

    get localId(): number;
    get remoteId(): number;
    get localCreated(): boolean;
    get serviceString(): string;
    get closed(): boolean;

    get onData(): Event<ArrayBuffer, AsyncEventResult>;
    get onClose(): Event<void, void>;

    constructor(controller: AdbSocketController);

    write(data: ArrayBuffer): Promise<void>;
    close(): Promise<void>;
}

It directly contains read events and write/close operations.

1.1 onData

onData is a special event emitter, it pauses the socket until first event handler is registered.

This is because I forgot that JavaScript is single-threaded, and worried about the dispatcher can read data from device and fire the event before user code can attach a listener.

1.2 onClose

onClose is also a special event emitter. After fired, newly attached event listeners will be invoked immediately (It latches the state).

1.3 write()

Simply write data to the other end.

2. Web Streams

Web Streams API defines standardized stream API for Web and Node.js.

It mainly contains three classes:

1. ReadableStream
2. WriteableStream
3. TransformStream

3. Proposed design

3.1 Basic

Because ReadableStream and WriteableStream are both classes, they can't be extended by a single class. So as the spec recommends, a duplex stream is implemented by an object with readable and writable properties:

class AdbSocket {
    public readonly readable: ReadableStream;
    public readonly writable: WriteableStream;
}

3.2 readable

readable ReadableStream will replace onData event and onClose event.

3.2.1 cancel

Calling cancel method on readable or one of its readers will close both readable and writable (and the socket itself).

3.2.2 Use patterns

onData event is a "push" model, while readable is a "pull" model. Converting to readable means changes to consumers.

Now onData is used in:

• AdbLegacyShell: It pipes onData events to its onStdout event:

  ya-webadb/libraries/adb/src/commands/shell/legacy.ts

  Line 34 in e6b751c

  this.socket.onData(this.handleData, this);

  ya-webadb/libraries/adb/src/commands/shell/legacy.ts

  Lines 38 to 41 in e6b751c

  	private handleData(data: ArrayBuffer) {
  	// Legacy shell doesn't support splitting output streams.
  	this.stdoutEvent.fire(data);
  	}

  After conversion: it will need a "pull loop".

• createSocketAndReadAll util method: Collects all data from a socket:

  ya-webadb/libraries/adb/src/adb.ts

  Lines 233 to 237 in 41a9565

  	let result = '';
  	socket.onData(buffer => {
  	result += decodeUtf8(buffer);
  	});
  	socket.onClose(() => resolver.resolve(result));

  After conversion: It can use readable as an async iterator. This will be simpler.

• AdbBufferedStream: Most consumers uses AdbBufferedStream to read exactly N bytes from the socket, ignoring packet boundaries.

  After conversion: AdbBufferedStream is still required, because ReadableStream doesn't support reading exactly N bytes.

3.3 writable

writable WritableStream will replace write method.

3.3.1 close

Calling close method on writable or one of its writers will close both readable and writeable (and the socket itself).

3.4 StructTransformStream

Web Streams support both byte stream and object stream. Struct deserialization can be wrapped in a TransformStream.

pseudo code:

const transformStream = new StructTransformStream(AdbPacket);
for await (const packet of transformStream) {
    // Dispatch packet
}

It also allows more separation between stream parsing and processing.

4. Advantages

• Web Streams enforce a single reader/writer model. Reduce possibility of mis-use.
• Improve interoperability with other libraries.

5. Disadvantages

• More complicate.

  • Now the consumer must maintain 5 objects (vs 1):

    1. AdbSocket
    2. readable
    3. reader
    4. writable
    5. writer

  • Incorrectly calling getReader/getWriter will lock the program.

  • It has 4 (or maybe 5) close methods (readable, reader, writable, writer, maybe also AdbSocket).

6. Unresolved Questions

6.1 Node.js doesn't expose Web Streams classes on global

In web browsers, ReadableStream and WriteableStream are exposed on Window.

In Node.js, they exist in stream/web module.

Possible solutions:

1. Create a helper package:

   1. Re-export them from stream/web for Node.js
   2. Read them from global for browsers
   3. Use browser field in package.json to let bundlers choose a different implementation.

   Downside: Requires a bundler, can't be used as native ESM in browsers.

2. Abuse the obsoleted commonjs module spec, with allows synchronized optional dependency
   let m; try { m = require("m") } catch {}
   Downside: Still doesn't work in native ESM, but requires less special treatment by bundlers.

3. Require users to re-assign them to global in Node.js

   Downside: Global pollution.

6.2 What should abort method on writable do?

[yume-chan]

It can use readable as an async iterator. This will be simpler.

1. Chrome doesn't support @@asyncIterator and values https://crbug.com/929585
2. Node.js v16.13.1 supports both https://nodejs.org/api/webstreams.html#readablestreamvaluesoptions
3. @types/node lacks the values method [@types/node] import('stream/web').ReadableStream missing a `values` method DefinitelyTyped/DefinitelyTyped#58791
4. @types/web have none of them (because no browsers support them) Support async iterable DOM types microsoft/TypeScript#29867

They can be polyfilled.

[yume-chan]

General guides:

1. Names of derive class of TransformStream only need to end with Stream, not TransformStream. For example TextDecoderStream in Web Encoding API.
2. Use highWaterMark and size on ReadableStream/WritableStreams to improve performance (otherwise it will always waiting data to arrive/be read).
3. Don't use highWaterMark and size on TransformStreams, otherwise there will be extra buffers in the pipe chain.
4. Run async code in start()/pull()/write() is fine, and throwing errors (including async errors) will error out the stream. Don't use try...catch... and manually call controller.error().
5. Always prefer await over returning a Promise. await clearly tells readers there is an async function.
6. Prefer pipeTo/pipeThrough over manually piping, to ensure back pressure works with minimal effort. A WrtaiableStream or identity TransformStream can be used to inspect a ReadableStream and do its process.

[yume-chan]

Open questions;

1. Events can be hooked up before initialization (e.g. connection), but streams are usually created after initialization (streams don't need to be hooked before initialization, because they have internal queue that buffers data), but it's harder to use. Maybe we need a proxy stream.
2. Some objects are re-usable, but streams need to be closed. It's conflicted with 1.
3. When will the streams close is extremely confusing.
4. When split one ReadableStream into multiple (for example one socket to multiple subprocess pipe), all streams must be consumed, or they will all stuck.

[yume-chan]

The object containing streams should be created by an async factory, and should be strictly single-used.

So the streams are ready once the object is created.

For example

class AdbWebUsbBackend {
    public connect(): { backend: this, readable: ReadableStream<ArrayBuffer>, writable: WritableStream<ArrayBuffer> } {
        if (this._connected) throw new Error('already connected');
    }
}

Which also means the caller of connect will "own" the streams, instead of anybody with access to backend can access the stream.

[yume-chan]

WebSocketStream proposal:

const wss = new WebSocketStream(WSS_URL);
const {readable, writable} = await wss.connection;

It starts connection in the constructor, then report the progress/result in the connection property.

However I can't use this style, because

1. Some objects like AdbBackend don't start connection in constructor, they have a dedicated connect method, so just return Promise<ReadableWritablePair> here is enough.
2. Some objects like AdbSubprocessProtocol are created by factory methods, they can't (hard to) be created using standalone constructors.
3. Some objects like AdbPacketDispatcher has too much methods to be returned from a Promise, it simply divides them to two objects (a factory and a real instance.

[yume-chan]

AdbBufferedStream is still required, because ReadableStream doesn't support reading exactly N bytes.

Upstream issues:

• Default stream:
  • Add support for bulk read operation whatwg/streams#1003
• BYOB
  • BYOB with "fill at least" semantics whatwg/streams#1175
  • ReadableStreamBYOBReader.prototype.readFully(view) whatwg/streams#1143
  • Add ReadableStreamBYOBReader.prototype.read(view, { min }) whatwg/streams#1145

However, whether it can be used with pipeTo is still unclear.

[yume-chan]

ADB Sync: wait vs error

Every calls to Adb.sync() returns a new AdbSync object that bound to separated AdbSocket.

Each AdbSync command consists of multiple request and response packets. After a command is successfully completed, the AdbSync object can start executing a new command. However, multiple commands can't be processed in parallel, because each packet doesn't identify which command it belongs to.

---

Before stream rewrite, each AdbSync object has a sendLock, which asynchronously blocks new commands while one is already being processed.

class AdbSync {
    socket: AdbSocket;
    sendLock: AutoResetEvent;

    async command1() {
        await this.sendLock.wait();

        try {
            // send/receive packets
            this.socket.write(new ArrayBuffer(0));
        } finally {
            this.sendLock.notify();
        }
    }
}

---

After stream rewrite, it's possible to re-use the lock mechanism of the AdbScoekt.writeable to prevent re-entrance, except, it will throw an error, instead of blocking.

class AdbSync {
    writable: WritableStream<ArrayBuffer>;

    async command1() {
        // Throw if stream is already locked
        const writer = this.writable.getWriter();

        try {
            // send/receive packets
            writer.write(new ArrayBuffer(0));
        } finally {
            writer.releaseLock();
        }
    }
}

In fact, because each command is implemented in their own methods, it can be more encapsulated.

function command1(writable: WritableStream<ArrayBuffer>) {
    const writer = writable.getWriter();

    try {
        // send/receive packets
        writer.write(new ArrayBuffer(0));
        return {};
    } finally {
        writer.releaseLock();
    }
}

class AdbSync {
    writable: WritableStream<ArrayBuffer>;

    async command1() {
        return command1(this.writable);
    }
}

---

Continuing using the sendLock is possible, but because it's very hard to observe internal state of streams, it now requires more care around "sync" methods that return a stream.

public write(
    filename: string,
    mode?: number,
    mtime?: number,
): WritableStream<ArrayBuffer> {
    return new HookWritableStream({
        start: async () => {
            await this.sendLock.wait();
            return {
                writable: adbSyncPush(
                    this.stream,
                    this.writer,
                    filename,
                    mode,
                    mtime,
                ),
                state: {},
            };
        },
        close: async () => {
            this.sendLock.notify();
        }
    });
}

On the other hand, it's not possible to get notified when a WritableStream became locked or unlocked.

[yume-chan]

Stream chunk type: ArrayBuffer vs Uint8Array?

Many streams defined in DOM specs are using Uint8Array as chunk type. Potentially for supporting "BYOB" (Bring Your Own Blob) streams.

Note that when they are used in BYOB mode, they can accept any ArrayBufferView, not only Uint8Array.

Maybe we should migrate all streams to use Uint8Array instead. Usually the performance is on par because both ArrayBuffer and Uint8Array are actually pointers.

[yume-chan]

Replace EventQueue

EventQueue is basically a simple ReadableStream that converts "push" system to "pull" system.

So ReadableStream can completely replaces EventQueue.

Replace AdbPacketDispatcher.onIncomingSocket?

WebTransport spec has such definition:

interface WebTransport {
    incomingBidirectionalStreams: ReadableStream<WebTransportBidirectionalStream>;
    incomingUnidirectionalStreams: ReadableStream<ReadableStream<Uint8Array>>;
}

So it uses ReadableStream to replace events. The advantage is already mentioned, ReadableStream will buffer the value until a reader read it, unlike an event will still fire and forget the value even if there is no event listener attached.

However, AdbPacketDispatcher.onIncomingSocket want to find exactly one handler that can handle the incoming socket (immediately, otherwise the whole connection will stale). This is not how ReadableStream works (can put a value back to front of queue). So it can't be replaced.

[yume-chan]

Vercel doesn't support Node.js 16

Vercel currently only support Node.js LTS 12 and 14, but stream/web requires 16.

It's possible to remove all usage of streams from Server side (they only make sense on client side, when a device is connected), but adds unnecessary complexity.

[yume-chan]

Specs using Streams API

• Web Streams API
  • Spec: https://streams.spec.whatwg.org/
  • GitHub: https://github.com/whatwg/streams
• Web Serial API
  • Spec: https://wicg.github.io/serial/
  • GitHub: https://github.com/WICG/serial
  • Most complete one, but has the wired "setting readable/writable to null after disconnection" behavior
• File System Access API
  • Spec: https://wicg.github.io/file-system-access/
  • GitHub: https://github.com/WICG/file-system-access
  • Doesn't use ReadableStream, only defines a (special) WritableStream (FileSystemWritableFileStream)
• File API
  • Spec: https://w3c.github.io/FileAPI/
  • GitHub: https://github.com/w3c/FileAPI
  • Defines one ReadableStream, only has placeholder algorithm.
• WebTransport API
  • Spec: https://www.w3.org/TR/webtransport/
  • GitHub: https://github.com/w3c/webtransport
  • Very unpolished, even didn't define how to close a stream
• Direct Sockets API
  • Spec: https://wicg.github.io/direct-sockets/
  • GitHub: https://github.com/WICG/direct-sockets
  • Only has some IDLs, no algorithms, no descriptions
• Encoding API
  • Spec: https://encoding.spec.whatwg.org/
  • GitHub: https://github.com/whatwg/encoding
  • Defines two TransformStreams
• Compression Streams API
  • Spec: https://wicg.github.io/compression
  • GitHub: https://github.com/WICG/compression
  • Defines two TransformStreams
• WebSocketStream API
  • Spec: None
  • GitHub: https://github.com/ricea/websocketstream-explainer
  • Has no spec