#1 support localized message handling for bi-directional streams

implements sendWithCallback so that a local function can handle a
message sent over a bi-directional stream. also, expose onError to the
client (rather than hard-end the websocket) to give the client better
control over how to handle errors

client/grpc-web/package.json

@@ -8,7 +8,8 @@
   "scripts": {
     "clean": "rm -rf dist",
     "postbootstrap": "npm run lib:build",
-    "lib:build": "npm run clean && webpack"
+    "lib:build": "npm run clean && webpack",
+    "lib:build-dev": "npm run clean && webpack -d --mode development"
   },
   "publishConfig": {
     "access": "public"

client/grpc-web/src/client.ts

@@ -19,12 +19,14 @@ export interface ClientRpcOptions extends RpcOptions {
 export interface Client<TRequest extends ProtobufMessage, TResponse extends ProtobufMessage> {
   start(metadata?: Metadata.ConstructorArg): void;
   send(message: TRequest): void;
+  sendWithCallback(message: TRequest, callback: (message: TResponse) => void): void;
   finishSend(): void;
   close(): void;
 
   onHeaders(callback: (headers: Metadata) => void): void;
   onMessage(callback: (message: TResponse) => void): void;
   onEnd(callback: (code: Code, message: string, trailers: Metadata) => void): void;
+  onError(callback: (code: Code, message: string, trailers: Metadata) => void): void;
 }
 
 export function client<TRequest extends ProtobufMessage, TResponse extends ProtobufMessage, M extends MethodDefinition<TRequest, TResponse>>(methodDescriptor: M, props: ClientRpcOptions): Client<TRequest, TResponse> {
@@ -44,6 +46,7 @@ class GrpcClient<TRequest extends ProtobufMessage, TResponse extends ProtobufMes
   onHeadersCallbacks: Array<(headers: Metadata) => void> = [];
   onMessageCallbacks: Array<(res: TResponse) => void> = [];
   onEndCallbacks: Array<(code: Code, message: string, trailers: Metadata) => void> = [];
+  onErrorCallbacks: Array<(code: Code, message: string, trailers: Metadata) => void> = [];
 
   transport: Transport;
   parser = new ChunkParser();
@@ -67,6 +70,7 @@ class GrpcClient<TRequest extends ProtobufMessage, TResponse extends ProtobufMes
       onHeaders: this.onTransportHeaders.bind(this),
       onChunk: this.onTransportChunk.bind(this),
       onEnd: this.onTransportEnd.bind(this),
+      onError: this.onTransportEnd.bind(this)
     };
 
     if (this.props.transport) {
@@ -228,9 +232,7 @@ class GrpcClient<TRequest extends ProtobufMessage, TResponse extends ProtobufMes
 
   rawOnError(code: Code, msg: string, trailers: Metadata = new Metadata()) {
     this.props.debug && debug("rawOnError", code, msg);
-    if (this.completed) return;
-    this.completed = true;
-    this.onEndCallbacks.forEach(callback => {
+    this.onErrorCallbacks.forEach(callback => {
       if (this.closed) return;
       try {
         callback(code, msg, trailers);
@@ -269,6 +271,10 @@ class GrpcClient<TRequest extends ProtobufMessage, TResponse extends ProtobufMes
     this.onEndCallbacks.push(callback);
   }
 
+  onError(callback: (code: Code, message: string, trailers: Metadata) => void) {
+    this.onErrorCallbacks.push(callback);
+  }
+
   start(metadata?: Metadata.ConstructorArg) {
     if (this.started) {
       throw new Error("Client already started - cannot .start()");
@@ -301,6 +307,12 @@ class GrpcClient<TRequest extends ProtobufMessage, TResponse extends ProtobufMes
     this.transport.sendMessage(msgBytes);
   }
 
+  sendWithCallback(message: TRequest, callback: (message: TResponse) => void) {
+    this.onMessageCallbacks = [];
+    this.onMessageCallbacks.push(callback);
+    this.send(message);
+  }
+
   finishSend() {
     if (!this.started) {
       throw new Error("Client not started - .finishSend() must be called before .close()");

client/grpc-web/src/transports/Transport.ts

@@ -27,6 +27,7 @@ export interface TransportOptions {
   onHeaders: (headers: Metadata, status: number) => void;
   onChunk: (chunkBytes: Uint8Array, flush?: boolean) => void;
   onEnd: (err?: Error) => void;
+  onError: (err?: Error) => void;
 }
 
 export interface TransportFactory {

client/grpc-web/src/transports/websocket/websocket.ts

@@ -72,6 +72,7 @@ function websocketRequest(options: TransportOptions): Transport {
 
       ws.onerror = function (error) {
         options.debug && debug("websocketRequest.onerror", error);
+        options.onError();
       };
 
       ws.onmessage = function (e) {