Add abort API support for promise calls

For grpc#1478

Author: sampajano

docker-compose.yml

@@ -1,4 +1,3 @@
-version: '3'
 services:
   prereqs:
     build:

javascript/net/grpc/web/abstractclientbase.js

@@ -32,6 +32,20 @@ const MethodDescriptor = goog.require('grpc.web.MethodDescriptor');
 const RpcError = goog.require('grpc.web.RpcError');
 
 
+/**
+ * @constructor
+ * @struct
+ * @final
+ */
+const PromiseCallOptions = function() {};
+
+/**
+ * An AbortSignal to abort the call.
+ * @type {AbortSignal|undefined}
+ */
+PromiseCallOptions.prototype.signal;
+
+
 /**
  * This interface represents a grpc-web client
  * @interface
@@ -62,10 +76,11 @@ const AbstractClientBase = class {
    * @param {!Object<string, string>} metadata User defined call metadata
    * @param {!MethodDescriptor<REQUEST, RESPONSE>}
    *   methodDescriptor Information of this RPC method
+   * @param options Options for the call
    * @return {!IThenable<RESPONSE>}
    *   A promise that resolves to the response message
    */
-  thenableCall(method, requestMessage, metadata, methodDescriptor) {}
+  thenableCall(method, requestMessage, metadata, methodDescriptor, options) {}
 
   /**
    * @abstract
@@ -78,21 +93,19 @@ const AbstractClientBase = class {
    * @return {!ClientReadableStream<RESPONSE>} The Client Readable Stream
    */
   serverStreaming(method, requestMessage, metadata, methodDescriptor) {}
-
-  /**
-   * Get the hostname of the current request.
-   * @static
-   * @template REQUEST, RESPONSE
-   * @param {string} method
-   * @param {!MethodDescriptor<REQUEST,RESPONSE>} methodDescriptor
-   * @return {string}
-   */
-  static getHostname(method, methodDescriptor) {
-    // method = hostname + methodDescriptor.name(relative path of this method)
-    return method.substr(0, method.length - methodDescriptor.name.length);
-  }
 };
 
+/**
+ * Get the hostname of the current request.
+ * @template REQUEST, RESPONSE
+ * @param {string} method
+ * @param {!MethodDescriptor<REQUEST,RESPONSE>} methodDescriptor
+ * @return {string}
+ */
+function getHostname(method, methodDescriptor) {
+  // method = hostname + methodDescriptor.name(relative path of this method)
+  return method.substr(0, method.length - methodDescriptor.name.length);
+}
 
 
-exports = AbstractClientBase;
+exports = {AbstractClientBase, PromiseCallOptions, getHostname};

javascript/net/grpc/web/grpcwebclientbase.js

@@ -28,7 +28,6 @@ goog.module('grpc.web.GrpcWebClientBase');
 goog.module.declareLegacyNamespace();
 
 
-const AbstractClientBase = goog.require('grpc.web.AbstractClientBase');
 const ClientOptions = goog.requireType('grpc.web.ClientOptions');
 const ClientReadableStream = goog.require('grpc.web.ClientReadableStream');
 const ClientUnaryCallImpl = goog.require('grpc.web.ClientUnaryCallImpl');
@@ -40,6 +39,7 @@ const RpcError = goog.require('grpc.web.RpcError');
 const StatusCode = goog.require('grpc.web.StatusCode');
 const XhrIo = goog.require('goog.net.XhrIo');
 const googCrypt = goog.require('goog.crypt.base64');
+const {AbstractClientBase, PromiseCallOptions, getHostname} = goog.require('grpc.web.AbstractClientBase');
 const {Status} = goog.require('grpc.web.Status');
 const {StreamInterceptor, UnaryInterceptor} = goog.require('grpc.web.Interceptor');
 const {toObject} = goog.require('goog.collections.maps');
@@ -101,7 +101,7 @@ class GrpcWebClientBase {
    * @export
    */
   rpcCall(method, requestMessage, metadata, methodDescriptor, callback) {
-    const hostname = AbstractClientBase.getHostname(method, methodDescriptor);
+    const hostname = getHostname(method, methodDescriptor);
     const invoker = GrpcWebClientBase.runInterceptors_(
         (request) => this.startStream_(request, hostname),
         this.streamInterceptors_);
@@ -112,18 +112,35 @@ class GrpcWebClientBase {
   }
 
   /**
-   * @override
-   * @export
+   * @param {string} method The method to invoke
+   * @param {REQUEST} requestMessage The request proto
+   * @param {!Object<string, string>} metadata User defined call metadata
+   * @param {!MethodDescriptor<REQUEST, RESPONSE>} methodDescriptor
+   * @param {?PromiseCallOptions=} options Options for the call
+   * @return {!Promise<RESPONSE>}
+   * @template REQUEST, RESPONSE
    */
-  thenableCall(method, requestMessage, metadata, methodDescriptor) {
-    const hostname = AbstractClientBase.getHostname(method, methodDescriptor);
+  thenableCall(
+      method, requestMessage, metadata, methodDescriptor, options = {}) {
+    const hostname = getHostname(method, methodDescriptor);
+    const signal = options && options.signal;
     const initialInvoker = (request) => new Promise((resolve, reject) => {
+      // If the signal is already aborted, immediately reject the promise
+      // and don't issue the call.
+      if (signal && signal.aborted) {
+        const error = new RpcError(StatusCode.CANCELLED, 'Aborted');
+        error.cause = signal.reason;
+        reject(error);
+        return;
+      }
+
       const stream = this.startStream_(request, hostname);
       let unaryMetadata;
       let unaryStatus;
       let unaryMsg;
       GrpcWebClientBase.setCallback_(
-          stream, (error, response, status, metadata, unaryResponseReceived) => {
+          stream,
+          (error, response, status, metadata, unaryResponseReceived) => {
             if (error) {
               reject(error);
             } else if (unaryResponseReceived) {
@@ -136,7 +153,19 @@ class GrpcWebClientBase {
               resolve(request.getMethodDescriptor().createUnaryResponse(
                   unaryMsg, unaryMetadata, unaryStatus));
             }
-          }, true);
+          },
+          true);
+
+      // Wire up cancellation from the abort signal, if any.
+      if (signal) {
+        signal.addEventListener('abort', () => {
+          stream.cancel();
+
+          const error = new RpcError(StatusCode.CANCELLED, 'Aborted');
+          error.cause = /** @type {!AbortSignal} */ (signal).reason;
+          reject(error);
+        });
+      }
     });
     const invoker = GrpcWebClientBase.runInterceptors_(
         initialInvoker, this.unaryInterceptors_);
@@ -152,20 +181,21 @@ class GrpcWebClientBase {
    * @param {!Object<string, string>} metadata User defined call metadata
    * @param {!MethodDescriptor<REQUEST, RESPONSE>} methodDescriptor Information
    *     of this RPC method
+   * @param {?PromiseCallOptions=} options Options for the call
    * @return {!Promise<RESPONSE>}
    * @template REQUEST, RESPONSE
    */
-  unaryCall(method, requestMessage, metadata, methodDescriptor) {
-    return /** @type {!Promise<RESPONSE>}*/ (
-        this.thenableCall(method, requestMessage, metadata, methodDescriptor));
+  unaryCall(method, requestMessage, metadata, methodDescriptor, options = {}) {
+    return /** @type {!Promise<RESPONSE>}*/ (this.thenableCall(
+        method, requestMessage, metadata, methodDescriptor, options));
   }
 
   /**
    * @override
    * @export
    */
   serverStreaming(method, requestMessage, metadata, methodDescriptor) {
-    const hostname = AbstractClientBase.getHostname(method, methodDescriptor);
+    const hostname = getHostname(method, methodDescriptor);
     const invoker = GrpcWebClientBase.runInterceptors_(
         (request) => this.startStream_(request, hostname),
         this.streamInterceptors_);
@@ -279,7 +309,9 @@ class GrpcWebClientBase {
             message: 'Incomplete response',
           });
         } else if (useUnaryResponse) {
-          callback(null, responseReceived, null, null, /* unaryResponseReceived= */ true);
+          callback(
+            null, responseReceived, null, null,
+            /* unaryResponseReceived= */ true);
         } else {
           callback(null, responseReceived);
         }
@@ -368,12 +400,9 @@ class GrpcWebClientBase {
    *     (!Promise<RESPONSE>|!ClientReadableStream<RESPONSE>)}
    */
   static runInterceptors_(invoker, interceptors) {
-    let curInvoker = invoker;
-    interceptors.forEach((interceptor) => {
-      const lastInvoker = curInvoker;
-      curInvoker = (request) => interceptor.intercept(request, lastInvoker);
-    });
-    return curInvoker;
+    return interceptors.reduce((accumulatedInvoker, interceptor) => {
+      return (request) => interceptor.intercept(request, accumulatedInvoker);
+    }, invoker);
   }
 }