Feature: support AsyncIterable instead of Observable

[paralin]

Currently the interface generated for an RPC is like:

export interface MyService {
  MyRpcCall(request: Observable<Packet>): Observable<Packet>
}

I am proposing to add a new option which instead generates:

import { Source, Sink } from 'it-stream-types'

export interface MyService {
  MyRpcCall(request: Source<Packet>): Source<Packet>
}

This would use it-pipe and the other iterables instead of Observable and rxjs.

[stephenh]

@paralin sure, that sgtm. Little worried about this becoming a slippery slope to supporting ~N different stream libraries, but understood that just is what it is for now, w/o something built-in to node.

[paralin]

@stephenh https://www.typescriptlang.org/docs/handbook/release-notes/typescript-2-3.html

As far as I know Async Iterators are actually somewhat more of a Javascript feature and closer to the "standard way" than rxjs Observables, since they're baked into the language.

The it-stream-types and it-pipe libraries are just nice interfaces on top to make it a bit easier.

[stephenh]

@paralin ah yeah, that's a good point! Even better.

Oh wow, yeah, that it-stream-types library is really small:

https://github.com/achingbrain/it-stream-types/blob/master/src/index.ts

Could we get by with just using AsyncIterator directly? Like if Source is just a type alias:

export type Source<T> = AsyncIterable<T> | Iterable<T>

I'm wondering if using AsyncIterable directly would still be just as ergonomic for users of it-pipe, without introducing that dependency in the ts-proto code itself?

[paralin]

Yes, I think the only difference is to replace the Observable type with AsyncIterable.

[stephenh]

Ah great! That sounds good to me. I think we have an useObservable flag already...iirc? If so, maybe a useObservable=rxjs / useObservable=async?

Fwiw given AsyncIterable is built-in, I'd be fine switching the default behavior to use it instead.

[paralin]

I've implemented the code generation part without adding any additional package deps.

The following examples have comments which indicate which it-stream-type they are compatible with.

Example of generated encodeTransform and decodeTransform functions:

  // encodeTransform encodes a source of message objects.
  // implements Transform<TestMessage, Uint8Array>
  // source: Source<TestMessage | TestMessage[]>
  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.
  // implements Transform<Uint8Array, TestMessage>
  // source: Source<Uint8Array | Uint8Array[]>
  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)]
      }
    }
  },

Example of the generated RPC for a bidirectional RPC:

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

I'm quite happy with how straightforward this ended up being to implement.

[paralin]

See #605

[stephenh]

🎉 This issue has been resolved in version 1.116.0 🎉

The release is available on:

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

Your semantic-release bot 📦🚀