-
Notifications
You must be signed in to change notification settings - Fork 337
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add option to use async iterables (#605)
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.
Fixes #600
Signed-off-by: Christian Stewart <christian@paral.in>- Loading branch information
Showing
20 changed files
with
248 additions
and
31 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| useAsyncIterable=true |
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| syntax = "proto3"; | ||
|
|
||
| package simple; | ||
|
|
||
| service Test { | ||
| rpc BidiStreaming(stream TestMessage) returns (stream TestMessage) {} | ||
| } | ||
|
|
||
| message TestMessage { | ||
| string value = 1; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,138 @@ | ||
| /* eslint-disable */ | ||
| import * as _m0 from 'protobufjs/minimal'; | ||
|
|
||
| export const protobufPackage = 'simple'; | ||
|
|
||
| export interface TestMessage { | ||
| value: string; | ||
| } | ||
|
|
||
| function createBaseTestMessage(): TestMessage { | ||
| return { value: '' }; | ||
| } | ||
|
|
||
| export const TestMessage = { | ||
| encode(message: TestMessage, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer { | ||
| if (message.value !== '') { | ||
| writer.uint32(10).string(message.value); | ||
| } | ||
| return writer; | ||
| }, | ||
|
|
||
| decode(input: _m0.Reader | Uint8Array, length?: number): TestMessage { | ||
| const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input); | ||
| let end = length === undefined ? reader.len : reader.pos + length; | ||
| const message = createBaseTestMessage(); | ||
| while (reader.pos < end) { | ||
| const tag = reader.uint32(); | ||
| switch (tag >>> 3) { | ||
| case 1: | ||
| message.value = reader.string(); | ||
| break; | ||
| default: | ||
| reader.skipType(tag & 7); | ||
| break; | ||
| } | ||
| } | ||
| return message; | ||
| }, | ||
|
|
||
| // 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)]; | ||
| } | ||
| } | ||
| }, | ||
|
|
||
| fromJSON(object: any): TestMessage { | ||
| return { | ||
| value: isSet(object.value) ? String(object.value) : '', | ||
| }; | ||
| }, | ||
|
|
||
| toJSON(message: TestMessage): unknown { | ||
| const obj: any = {}; | ||
| message.value !== undefined && (obj.value = message.value); | ||
| return obj; | ||
| }, | ||
|
|
||
| fromPartial<I extends Exact<DeepPartial<TestMessage>, I>>(object: I): TestMessage { | ||
| const message = createBaseTestMessage(); | ||
| message.value = object.value ?? ''; | ||
| return message; | ||
| }, | ||
| }; | ||
|
|
||
| export interface Test { | ||
| BidiStreaming(request: AsyncIterable<TestMessage>): AsyncIterable<TestMessage>; | ||
| } | ||
|
|
||
| export class TestClientImpl implements Test { | ||
| private readonly rpc: Rpc; | ||
| constructor(rpc: Rpc) { | ||
| this.rpc = rpc; | ||
| this.BidiStreaming = this.BidiStreaming.bind(this); | ||
| } | ||
| BidiStreaming(request: AsyncIterable<TestMessage>): AsyncIterable<TestMessage> { | ||
| const data = TestMessage.encodeTransform(request); | ||
| const result = this.rpc.bidirectionalStreamingRequest('simple.Test', 'BidiStreaming', data); | ||
| return TestMessage.decodeTransform(result); | ||
| } | ||
| } | ||
|
|
||
| interface Rpc { | ||
| request(service: string, method: string, data: Uint8Array): Promise<Uint8Array>; | ||
| clientStreamingRequest(service: string, method: string, data: AsyncIterable<Uint8Array>): Promise<Uint8Array>; | ||
| serverStreamingRequest(service: string, method: string, data: Uint8Array): AsyncIterable<Uint8Array>; | ||
| bidirectionalStreamingRequest( | ||
| service: string, | ||
| method: string, | ||
| data: AsyncIterable<Uint8Array> | ||
| ): AsyncIterable<Uint8Array>; | ||
| } | ||
|
|
||
| type Builtin = Date | Function | Uint8Array | string | number | boolean | undefined; | ||
|
|
||
| export type DeepPartial<T> = T extends Builtin | ||
| ? T | ||
| : T extends Array<infer U> | ||
| ? Array<DeepPartial<U>> | ||
| : T extends ReadonlyArray<infer U> | ||
| ? ReadonlyArray<DeepPartial<U>> | ||
| : T extends {} | ||
| ? { [K in keyof T]?: DeepPartial<T[K]> } | ||
| : Partial<T>; | ||
|
|
||
| type KeysOfUnion<T> = T extends T ? keyof T : never; | ||
| export type Exact<P, I extends P> = P extends Builtin | ||
| ? P | ||
| : P & { [K in keyof P]: Exact<P[K], I[K]> } & Record<Exclude<keyof I, KeysOfUnion<P>>, never>; | ||
|
|
||
| function isSet(value: any): boolean { | ||
| return value !== null && value !== undefined; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| import { code, Code } from 'ts-poet'; | ||
|
|
||
| /** Creates a function to transform a message Source to a Uint8Array Source. */ | ||
| export function generateEncodeTransform(fullName: string): Code { | ||
| return code` | ||
| // encodeTransform encodes a source of message objects. | ||
| // Transform<${fullName}, Uint8Array> | ||
| async *encodeTransform( | ||
| source: AsyncIterable<${fullName} | ${fullName}[]> | Iterable<${fullName} | ${fullName}[]> | ||
| ): AsyncIterable<Uint8Array> { | ||
| for await (const pkt of source) { | ||
| if (Array.isArray(pkt)) { | ||
| for (const p of pkt) { | ||
| yield* [${fullName}.encode(p).finish()] | ||
| } | ||
| } else { | ||
| yield* [${fullName}.encode(pkt).finish()] | ||
| } | ||
| } | ||
| } | ||
| `; | ||
| } | ||
|
|
||
| /** Creates a function to transform a Uint8Array Source to a message Source. */ | ||
| export function generateDecodeTransform(fullName: string): Code { | ||
| return code` | ||
| // decodeTransform decodes a source of encoded messages. | ||
| // Transform<Uint8Array, ${fullName}> | ||
| async *decodeTransform( | ||
| source: AsyncIterable<Uint8Array | Uint8Array[]> | Iterable<Uint8Array | Uint8Array[]> | ||
| ): AsyncIterable<${fullName}> { | ||
| for await (const pkt of source) { | ||
| if (Array.isArray(pkt)) { | ||
| for (const p of pkt) { | ||
| yield* [${fullName}.decode(p)] | ||
| } | ||
| } else { | ||
| yield* [${fullName}.decode(pkt)] | ||
| } | ||
| } | ||
| } | ||
| `; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.