Support more gRPC streaming events. (#68)

- 'metadata' and 'status' events are meaningful and should be
  forwarded.
- gRPC streaming results have 'cancel' method, this method call
  needs to be forwarded as well.

Fixes #67

Author: jmuk

packages/packages/google-gax/lib/streaming.js

@@ -70,6 +70,7 @@ function StreamProxy(type, callback) {
   this._callback = callback;
   this._writeQueue = [];
   this._isEndCalled = false;
+  this._isCancelCalled = false;
   this.on('finish', this._onFinish.bind(this));
 }
 
@@ -102,6 +103,14 @@ StreamProxy.prototype._onFinish = function() {
   }
 };
 
+StreamProxy.prototype.cancel = function() {
+  if (this.stream) {
+    this.stream.cancel();
+  } else {
+    this._isCancelCalled = true;
+  }
+};
+
 /**
  * Specifies the target stream.
  * @param {ApiCall} apiCall - the API function to be called.
@@ -110,11 +119,30 @@ StreamProxy.prototype._onFinish = function() {
 StreamProxy.prototype.setStream = function(apiCall, argument) {
   var stream = apiCall(argument, this._callback);
   this.stream = stream;
-  stream.on('error', this.emit.bind(this, 'error'));
+  (['error', 'metadata', 'status']).forEach(function(event) {
+    stream.on(event, this.emit.bind(this, event));
+  }.bind(this));
+  // We also want to supply the status data as 'response' event to support
+  // the behavior of google-cloud-node expects.
+  // see: https://github.com/GoogleCloudPlatform/google-cloud-node/pull/1775#issuecomment-259141029
+  // https://github.com/GoogleCloudPlatform/google-cloud-node/blob/116436fa789d8b0f7fc5100b19b424e3ec63e6bf/packages/common/src/grpc-service.js#L355
+  stream.on('metadata', function(metadata) {
+    // Create a response object with succeeds.
+    // TODO: unify this logic with the decoration of gRPC response when it's added.
+    // see: https://github.com/googleapis/gax-nodejs/issues/65
+    this.emit('response', {
+      code: 200,
+      details: '',
+      message: 'OK',
+      metadata: metadata
+    });
+  }.bind(this));
   if (this.type !== StreamType.CLIENT_STREAMING) {
     stream.on('data', this.emit.bind(this, 'data'));
     // Pushing null causes an ending process of the readable stream.
     stream.on('end', this.push.bind(this, null));
+    // This is required in case no 'data' handler exists.
+    this.resume();
   }
   if (this.type !== StreamType.SERVER_STREAMING) {
     this._writeQueue.forEach(function(data) {
@@ -124,6 +152,9 @@ StreamProxy.prototype.setStream = function(apiCall, argument) {
     if (this._isEndCalled) {
       stream.end();
     }
+    if (this._isCancelCalled) {
+      stream.cancel();
+    }
   }
 };
 

packages/packages/google-gax/package.json

@@ -1,6 +1,6 @@
 {
   "name": "google-gax",
-  "version": "0.9.0",
+  "version": "0.9.1",
   "description": "Google API Extensions",
   "main": "index.js",
   "files": [

packages/packages/google-gax/test/api_callable.js

@@ -669,4 +669,85 @@ describe('streaming', function() {
     s.write(arg);
     s.end();
   });
+
+  it('forwards metadata and status', function(done) {
+    var responseMetadata = {metadata: true};
+    var status = {code: 0, metadata: responseMetadata};
+    var expectedResponse = {
+      code: 200,
+      message: 'OK',
+      details: '',
+      metadata: responseMetadata
+    };
+    function func(metadata, options) {
+      var s = through2.obj();
+      setTimeout(s.emit.bind(s, 'metadata', responseMetadata), 10);
+      s.on('end', s.emit.bind(s, 'status', status));
+      return s;
+    }
+    var apiCall = createStreamingCall(
+        func, streaming.StreamType.BIDI_STREAMING);
+    var s = apiCall(null, null);
+    var receivedMetadata;
+    var receivedStatus;
+    var receivedResponse;
+    s.on('metadata', function(data) {
+      receivedMetadata = data;
+    });
+    s.on('status', function(data) {
+      receivedStatus = data;
+    });
+    s.on('response', function(data) {
+      receivedResponse = data;
+    });
+    s.on('end', function() {
+      expect(receivedMetadata).to.deep.eq(responseMetadata);
+      expect(receivedStatus).to.deep.eq(status);
+      expect(receivedResponse).to.deep.eq(expectedResponse);
+      done();
+    });
+    expect(s.readable).to.be.true;
+    expect(s.writable).to.be.true;
+    setTimeout(s.end.bind(s), 50);
+  });
+
+  it('cancels in the middle', function(done) {
+    function schedulePush(s, c) {
+      if (!s.readable) {
+        return;
+      }
+      setTimeout(function() {
+        s.push(c);
+        schedulePush(s, c + 1);
+      }, 10);
+    }
+    var cancelError = new Error('cancelled');
+    function func(metadata, options) {
+      var s = through2.obj();
+      schedulePush(s, 0);
+      s.cancel = function() {
+        s.end();
+        s.emit('error', cancelError);
+      };
+      return s;
+    }
+    var apiCall = createStreamingCall(
+        func, streaming.StreamType.SERVER_STREAMING);
+    var s = apiCall(null, null);
+    var counter = 0;
+    var expectedCount = 5;
+    s.on('data', function(data) {
+      expect(data).to.eq(counter);
+      counter++;
+      if (counter === expectedCount) {
+        s.cancel();
+      } else if (counter > expectedCount) {
+        done(new Error('should not reach'));
+      }
+    });
+    s.on('error', function(err) {
+      expect(err).to.eq(cancelError);
+      done();
+    });
+  });
 });