grpc-web/javascript/net/grpc/web/streambodyclientreadablestream.js

/**
 *
 * Copyright 2018 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
/**
 * @fileoverview gRPC web client Readable Stream
 *
 * This class is being returned after a gRPC streaming call has been
 * started. This class provides functionality for user to operates on
 * the stream, e.g. set onData callback, etc.
 *
 * This wraps the underlying goog.net.streams.NodeReadableStream
 *
 * @author stanleycheung@google.com (Stanley Cheung)
 */
goog.module('grpc.web.StreamBodyClientReadableStream');

goog.module.declareLegacyNamespace();


const ClientReadableStream = goog.require('grpc.web.ClientReadableStream');
const ErrorCode = goog.require('goog.net.ErrorCode');
const EventType = goog.require('goog.net.EventType');
const GoogleRpcStatus = goog.require('proto.google.rpc.Status');
const GrpcWebError = goog.requireType('grpc.web.Error');
const Metadata = goog.requireType('grpc.web.Metadata');
const NodeReadableStream = goog.require('goog.net.streams.NodeReadableStream');
const StatusCode = goog.require('grpc.web.StatusCode');
const XhrIo = goog.require('goog.net.XhrIo');
const asserts = goog.require('goog.asserts');
const events = goog.require('goog.events');
const {GenericTransportInterface} = goog.require('grpc.web.GenericTransportInterface');
const {Status} = goog.require('grpc.web.Status');



/**
 * A stream that the client can read from. Used for calls that are streaming
 * from the server side.
 * @template RESPONSE
 * @implements {ClientReadableStream}
 * @final
 * @unrestricted
 */
class StreamBodyClientReadableStream {
  /**
   * @param {!GenericTransportInterface} genericTransportInterface
   * @param {function(?): RESPONSE} responseDeserializeFn
   * @param {boolean} isBinary
   */
  constructor(genericTransportInterface, responseDeserializeFn, isBinary) {
    /**
     * Whether or not the response protobuffer format is binary.
     * @private
     * @const {boolean}
     */
    this.isBinary_ = isBinary;

    /**
     * @private
     * @const {?NodeReadableStream|undefined} The XHR Node Readable Stream
     */
    this.xhrNodeReadableStream_ = genericTransportInterface.nodeReadableStream;

    /**
     * @private
     * @const {function(?): RESPONSE} The deserialize function for the proto
     */
    this.grpcResponseDeserializeFn_ = responseDeserializeFn;

    /**
     * @private
     * @const {?XhrIo|undefined} The XhrIo object
     */
    this.xhr_ = genericTransportInterface.xhr;

    /**
     * @private
     * @const {!Array<function(RESPONSE)>} The list of data callback
     */
    this.onDataCallbacks_ = [];

    /**
     * @private
     * @const {!Array<function(!Metadata)>} The list of metadata callbacks
     */
    this.onMetadataCallbacks_ = [];

    /**
     * @private
     * @const {!Array<function(!Status)>} The list of status callback
     */
    this.onStatusCallbacks_ = [];

    /**
     * @private
     * @const {!Array<function(...)>} The list of stream end callback
     */
    this.onEndCallbacks_ = [];

    /**
     * @private
     * @const {!Array<function(!GrpcWebError)>} The list of error callback
     */
    this.onErrorCallbacks_ = [];

    if (this.xhrNodeReadableStream_) {
      this.setStreamCallback_();
    }
  }


  /**
   * Set up the callback functions for unary calls.
   * @param {boolean} base64Encoded True if
   *     'X-Goog-Encode-Response-If-Executable' is 'base64' in request headers.
   */
  setUnaryCallback(base64Encoded) {
    events.listen(/** @type {!XhrIo}*/ (this.xhr_), EventType.COMPLETE, (e) => {
      if (this.xhr_.isSuccess()) {
        let response;
        if (this.isBinary_) {
          response = this.decodeBinaryResponse_(base64Encoded);
        } else {
          response = this.decodeJspbResponse_(base64Encoded);
        }
        const responseMessage = this.grpcResponseDeserializeFn_(response);
        const grpcStatus = StatusCode.fromHttpStatus(this.xhr_.getStatus());
        this.sendMetadataCallbacks_(this.readHeaders_());
        if (grpcStatus == StatusCode.OK) {
          this.sendDataCallbacks_(responseMessage);
        } else {
          this.sendErrorCallbacks_(
              /** @type {!GrpcWebError} */ (
                  {code: grpcStatus, message: response}));
        }
      } else {
        let rawResponse;
        if (this.isBinary_) {
          const xhrResponse = this.xhr_.getResponse();
          if (xhrResponse) {
            rawResponse =
                new Uint8Array(/** @type {!ArrayBuffer} */ (xhrResponse));
          }
        } else {
          rawResponse = this.xhr_.getResponseText();
        }

        let code;
        let message;
        let metadata = {};
        if (rawResponse) {
          const status = this.parseRpcStatus_(rawResponse);
          code = status.code;
          message = status.details;
          metadata = status.metadata;
        } else {
          code = StatusCode.UNKNOWN;
          message = 'Rpc failed due to xhr error. error code: ' +
              this.xhr_.getLastErrorCode() +
              ', error: ' + this.xhr_.getLastError();
        }
        this.sendMetadataCallbacks_(this.readHeaders_());
        this.sendErrorCallbacks_({code, message, metadata});
      }
    });
  }

  /**
   * @private
   */
  setStreamCallback_() {
    // Add the callback to the underlying stream
    this.xhrNodeReadableStream_.on('data', (data) => {
      if ('1' in data) {
        const response = this.grpcResponseDeserializeFn_(data['1']);
        this.sendDataCallbacks_(response);
      }
      if ('2' in data) {
        const status = this.parseRpcStatus_(data['2']);
        this.sendStatusCallbacks_(status);
      }
    });
    this.xhrNodeReadableStream_.on('end', () => {
      this.sendMetadataCallbacks_(this.readHeaders_());
      this.sendEndCallbacks_();
    });
    this.xhrNodeReadableStream_.on('error', () => {
      if (this.onErrorCallbacks_.length == 0) return;
      let lastErrorCode = this.xhr_.getLastErrorCode();
      if (lastErrorCode === ErrorCode.NO_ERROR && !this.xhr_.isSuccess()) {
        // The lastErrorCode on the XHR isn't useful in this case, but the XHR
        // status is. Full details about the failure should be available in the
        // status handler.
        lastErrorCode = ErrorCode.HTTP_ERROR;
      }

      let grpcStatusCode;
      switch (lastErrorCode) {
        case ErrorCode.NO_ERROR:
          grpcStatusCode = StatusCode.UNKNOWN;
          break;
        case ErrorCode.ABORT:
          grpcStatusCode = StatusCode.ABORTED;
          break;
        case ErrorCode.TIMEOUT:
          grpcStatusCode = StatusCode.DEADLINE_EXCEEDED;
          break;
        case ErrorCode.HTTP_ERROR:
          grpcStatusCode = StatusCode.fromHttpStatus(this.xhr_.getStatus());
          break;
        default:
          grpcStatusCode = StatusCode.UNAVAILABLE;
      }

      this.sendErrorCallbacks_({
        code: grpcStatusCode,
        // TODO(armiller): get the message from the response?
        // GoogleRpcStatus.deserialize(rawResponse).getMessage()?
        // Perhaps do the same status logic as in on('data') above?
        message: ErrorCode.getDebugMessage(lastErrorCode)
      });
    });
  }

  /**
   * @private
   * @param {boolean} base64Encoded
   * @return {string}
   */
  decodeJspbResponse_(base64Encoded) {
    // If the response is serialized as Base64 (for example if the
    // X-Goog-Encode-Response-If-Executable header is in effect), decode it
    // before passing it to the deserializer.
    let responseText = this.xhr_.getResponseText();
    if (base64Encoded &&
        this.xhr_.getResponseHeader(XhrIo.CONTENT_TYPE_HEADER) ===
            'text/plain') {
      if (!atob) {
        throw new Error('Cannot decode Base64 response');
      }
      responseText = atob(responseText);
    }
    return responseText;
  }

  /**
   * @private
   * @param {boolean} base64Encoded
   * @return {*}
   */
  decodeBinaryResponse_(base64Encoded) {
    if (base64Encoded &&
        this.xhr_.getResponseHeader('X-Goog-Safety-Encoding') == 'base64') {
      // Convert the response's ArrayBuffer to a string, which should
      // be a base64 encoded string.
      const bytes = new Uint8Array(
          /** @type {!ArrayBuffer} */ (this.xhr_.getResponse()));
      let byteSource = '';
      for (let i = 0; i < bytes.length; i++) {
        byteSource += String.fromCharCode(bytes[i]);
      }
      return byteSource;
    } else {
      return this.xhr_.getResponse();
    }
  }

  /**
   * @private
   * @return {!Metadata}
   */
  readHeaders_() {
    const initialMetadata = {};
    const responseHeaders = this.xhr_.getResponseHeaders();
    Object.keys(responseHeaders).forEach((header) => {
      initialMetadata[header] = responseHeaders[header];
    });
    return initialMetadata;
  }

  /**
   * @private
   * @param {!Uint8Array|string} data Data returned from the underlying stream.
   * @return {!Status} The Rpc Status details.
   */
  parseRpcStatus_(data) {
    let code;
    let message;
    const metadata = {};
    try {
      let rpcStatus;
      if (this.isBinary_) {
        rpcStatus = GoogleRpcStatus.deserializeBinary(data);
      } else {
        asserts.assertString(
            data, 'RPC status must be string in gRPC-Web jspb mode.');
        rpcStatus = GoogleRpcStatus.deserialize(data);
      }
      code = rpcStatus.getCode();
      message = rpcStatus.getMessage();
      if (rpcStatus.getDetailsList().length) {
        metadata['grpc-web-status-details-bin'] = data;
      }
    } catch (e) {
      // 404s may be accompanied by a GoogleRpcStatus. If they are not,
      // the generic message will fail to deserialize because it is not a
      // status.
      if (this.xhr_ && this.xhr_.getStatus() === 404) {
        code = StatusCode.NOT_FOUND;
        message = 'Not Found: ' + this.xhr_.getLastUri();
      } else {
        code = StatusCode.UNAVAILABLE;
        message = 'Unable to parse RpcStatus: ' + e;
      }
    }
    const status = {code: code, details: message, metadata: metadata};
    return status;
  }

  /**
   * @override
   * @export
   */
  on(eventType, callback) {
    // TODO(stanleycheung): change eventType to @enum type
    if (eventType == 'data') {
      this.onDataCallbacks_.push(callback);
    } else if (eventType == 'metadata') {
      this.onMetadataCallbacks_.push(callback);
    } else if (eventType == 'status') {
      this.onStatusCallbacks_.push(callback);
    } else if (eventType == 'end') {
      this.onEndCallbacks_.push(callback);
    } else if (eventType == 'error') {
      this.onErrorCallbacks_.push(callback);
    }
    return this;
  }

  /**
   * @private
   * @param {!Array<function(?)>} callbacks the internal list of callbacks
   * @param {function(?)} callback the callback to remove
   */
  removeListenerFromCallbacks_(callbacks, callback) {
    const index = callbacks.indexOf(callback);
    if (index > -1) {
      callbacks.splice(index, 1);
    }
  }

  /**
   * @export
   * @override
   */
  removeListener(eventType, callback) {
    if (eventType == 'data') {
      this.removeListenerFromCallbacks_(this.onDataCallbacks_, callback);
    } else if (eventType == 'metadata') {
      this.removeListenerFromCallbacks_(this.onMetadataCallbacks_, callback);
    } else if (eventType == 'status') {
      this.removeListenerFromCallbacks_(this.onStatusCallbacks_, callback);
    } else if (eventType == 'end') {
      this.removeListenerFromCallbacks_(this.onEndCallbacks_, callback);
    } else if (eventType == 'error') {
      this.removeListenerFromCallbacks_(this.onErrorCallbacks_, callback);
    }
    return this;
  }

  /**
   * @override
   * @export
   */
  cancel() {
    this.xhr_.abort();
  }

  /**
   * @private
   * @param {!RESPONSE} data The data to send back
   */
  sendDataCallbacks_(data) {
    for (let i = 0; i < this.onDataCallbacks_.length; i++) {
      this.onDataCallbacks_[i](data);
    }
  }

  /**
   * @private
   * @param {!Metadata} metadata The metadata to send back
   */
  sendMetadataCallbacks_(metadata) {
    for (let i = 0; i < this.onMetadataCallbacks_.length; i++) {
      this.onMetadataCallbacks_[i](metadata);
    }
  }

  /**
   * @private
   * @param {!Status} status The status to send back
   */
  sendStatusCallbacks_(status) {
    for (let i = 0; i < this.onStatusCallbacks_.length; i++) {
      this.onStatusCallbacks_[i](status);
    }
  }

  /**
   * @private
   * @param {!GrpcWebError} error The error to send back
   */
  sendErrorCallbacks_(error) {
    for (let i = 0; i < this.onErrorCallbacks_.length; i++) {
      this.onErrorCallbacks_[i](error);
    }
  }

  /**
   * @private
   */
  sendEndCallbacks_() {
    for (let i = 0; i < this.onEndCallbacks_.length; i++) {
      this.onEndCallbacks_[i]();
    }
  }
}



exports = StreamBodyClientReadableStream;