feat: add option to use async iterables

[paralin]

Adds option useAsyncIterable which uses AsyncIterable instead of Observable.

For example:

  bidirectionalStreamingRequest(
    service: string,
    method: string,
    data: AsyncIterable<Uint8Array>
  ): AsyncIterable<Uint8Array>

Generates Transform async iterables for encoding and decoding:

  // encodeTransform encodes a source of message objects.
  // Transform<TestMessage, Uint8Array>
  async *encodeTransform(
    source: AsyncIterable<TestMessage | TestMessage[]> | Iterable<TestMessage | TestMessage[]>
  ): AsyncIterable<Uint8Array> {
    for await (const pkt of source) {
      if (Array.isArray(pkt)) {
        for (const p of pkt) {
          yield* [TestMessage.encode(p).finish()];
        }
      } else {
        yield* [TestMessage.encode(pkt).finish()];
      }
    }
  },

  // decodeTransform decodes a source of encoded messages.
  // Transform<Uint8Array, TestMessage>
  async *decodeTransform(
    source: AsyncIterable<Uint8Array | Uint8Array[]> | Iterable<Uint8Array | Uint8Array[]>
  ): AsyncIterable<TestMessage> {
    for await (const pkt of source) {
      if (Array.isArray(pkt)) {
        for (const p of pkt) {
          yield* [TestMessage.decode(p)];
        }
      } else {
        yield* [TestMessage.decode(pkt)];
      }
    }
  },

Generates RPC service implementations which use the Transform iterators:

  BidiStreaming(request: AsyncIterable<TestMessage>): AsyncIterable<TestMessage> {
    const data = TestMessage.encodeTransform(request);
    const result = this.rpc.bidirectionalStreamingRequest('simple.Test', 'BidiStreaming', data);
    return TestMessage.decodeTransform(result);
  }

AsyncIterables indicate a stream has ended by closing with an optional error.

[paralin]

Note: I avoided using it-stream-types to avoid the extra dependency:

export type Source<T> = AsyncIterable<T> | Iterable<T>
source: Source<Uint8Array | Uint8Array[]>

=>

source: AsyncIterable<Uint8Array | Uint8Array[]> | Iterable<Uint8Array | Uint8Array[]>

[paralin]

@stephenh Tested this & released for starpc, converted everything to AsyncIterable from Observable: aperturerobotics/starpc@4ecfbe9

[paralin]

@stephenh This is ready to go now that #603 is merged

[stephenh]

Looks great, thanks!

[stephenh]

🎉 This PR is included in version 1.116.0 🎉

The release is available on:

• npm package (@latest dist-tag)
• GitHub release
• v1.116.0

Your semantic-release bot 📦🚀

[paralin]

Awesome, thanks @stephenh !