Streaming support

[Jille]

Streaming support currently seems broken. It generates a file with parse errors for me at least. It looks like there is partial support, going in the direction of using rxjs Observables.

Would it be valuable to agree on the interface of where we(/you) want to go?

Is there a plan for cancellation of streaming RPCs?

@@ -2,6 +2,7 @@
 import Long from 'long';
 import _m0 from 'protobufjs/minimal';
 import { Observable } from 'rxjs';
+import { map } from 'rxjs/operators';
 import { Timestamp } from './google/protobuf/timestamp';
 
 export const protobufPackage = '';
@@ @@
   
     MyServerStreaming(
       request: MyServerStreamingRequest
-    ): Promise<MyServerStreamingResponse> {
+    ): Observable<MyServerStreamingResponse> {
       const data = MyServerStreamingRequest.encode(request).finish();
-      const promise = this.rpc.request(
+      const stream = this.rpc.serverStreaming(
         
         "MyService",
         "MyServerStreaming",
         data
       );
-      return promise.then(data => MyServerStreamingResponse.decode(new _m0.Reader(data)));
+      return stream.pipe(map(packet => MyServerStreamingResponse.decode(new _m0.Reader(packet))));
     }
   
     MyBiDi(
       request: Observable<MyBiDiRequest>
-    ): Promise<MyBiDiResponse> {
-      const data = Observable<MyBiDiRequest>.encode(request).finish();
-      const promise = this.rpc.request(
+    ): Observable<MyBiDiResponse> {
+      const data = request.pipe(map(packet => MyBiDiRequest.encode(packet).finish()));
+      const stream = this.rpc.bidirectionalStreaming(
         
         "MyxService",
         "MyBiDi",
         data
       );
-      return promise.then(data => MyBiDiResponse.decode(new _m0.Reader(data)));
+      return stream.pipe(map(packet => MyBiDiResponse.decode(new _m0.Reader(packet))));
     }
   
     GetProduct(
@@  @@
 interface Rpc {
       request(
         
         service: string,
         method: string,
         data: Uint8Array
       ): Promise<Uint8Array>;
+
+      serverStreaming(
+        
+        service: string,
+        method: string,
+        data: Uint8Array
+      ): Observable<Uint8Array>;
+
+      clientStreaming(
+        
+        service: string,
+        method: string,
+        data: Observable<Uint8Array>
+      ): Uint8Array;
+
+      bidirectionalStreaming(
+        
+        service: string,
+        method: string,
+        data: Observable<Uint8Array>
+      ): Observable<Uint8Array>;
+    }
 
 declare var self: any | undefined;

We could conditionally add those extra methods to the Rpc interface iff there is a single streaming RPC? Given that it currently doesn't compile if people have a streaming RPC defined, that shouldn't break anyone?

[Jille]

So I went ahead and implemented this (PR 373).

Do we want to require all methods (serverStreamingRequest, clientStreamingRequest, bidirectionalStreamingRequest) if the user has a single streaming RPC, or do we want to expose all 4 methods iff they're used?

[stephenh]

@Jille I like what you did in the PR, re conditionally including the additional methods. If anything it's nice for backwards compatibility (although I'm also fine with breaking changes if necessary, but yeah if we can avoid it). Thanks!

[Jille]

PR 373 was merged :)