fix: Parse response errors and emit as error event. (#172)

package.json

@@ -76,6 +76,7 @@
     "test": "npm run cover"
   },
   "dependencies": {
+    "@google-cloud/common": "^0.24.0",
     "extend": "^3.0.1",
     "google-gax": "^0.20.0",
     "google-proto-files": "^0.16.0",
@@ -100,6 +101,7 @@
     "nyc": "^13.0.0",
     "power-assert": "^1.6.0",
     "prettier": "^1.13.5",
+    "proxyquire": "^2.1.0",
     "safe-buffer": "^5.1.2",
     "sinon": "^6.0.0"
   }

src/helpers.js

@@ -16,6 +16,7 @@
 
 'use strict';
 
+var common = require('@google-cloud/common');
 var pumpify = require('pumpify');
 var streamEvents = require('stream-events');
 var through = require('through2');
@@ -108,7 +109,14 @@ module.exports = () => {
           next(null, payload);
         }),
         requestStream,
-        through.obj(),
+        through.obj((response, enc, next) => {
+          if (response.error) {
+            next(new common.util.ApiError(response.error));
+            return;
+          }
+
+          next(null, response);
+        }),
       ]);
     });
 

test/helpers.test.js

@@ -18,14 +18,38 @@
 
 var assert = require('assert');
 var Buffer = require('safe-buffer').Buffer;
+var common = require('@google-cloud/common');
+var proxyquire = require('proxyquire');
 var sinon = require('sinon');
 var stream = require('stream');
 
-var speech = require('../');
-
 describe('Speech helper methods', () => {
-  var sandbox = sinon.sandbox.create();
   var client;
+  var FakeApiErrorOverride;
+  var sandbox = sinon.createSandbox();
+  var speech;
+
+  class FakeApiError extends common.util.ApiError {
+    constructor(error) {
+      super();
+
+      if (FakeApiErrorOverride) {
+        return FakeApiErrorOverride(error);
+      }
+    }
+  }
+
+  before(() => {
+    speech = proxyquire('../', {
+      './helpers.js': proxyquire('../src/helpers.js', {
+        '@google-cloud/common': {
+          util: {
+            ApiError: FakeApiError,
+          },
+        },
+      }),
+    });
+  });
 
   beforeEach(() => {
     client = new speech.v1.SpeechClient();
@@ -105,6 +129,33 @@ describe('Speech helper methods', () => {
       requestStream.emit('error', error);
     });
 
+    it('destroys the user stream when the response contains an error', done => {
+      // Stub the underlying _streamingRecognize method to just return
+      // a bogus stream.
+      var requestStream = new stream.PassThrough({objectMode: true});
+      sandbox
+        .stub(client._innerApiCalls, 'streamingRecognize')
+        .returns(requestStream);
+
+      var userStream = client.streamingRecognize(CONFIG, OPTIONS);
+
+      var error = {};
+      var fakeApiError = {};
+
+      FakeApiErrorOverride = err => {
+        assert.strictEqual(err, error);
+        return fakeApiError;
+      };
+
+      userStream.once('error', err => {
+        assert.strictEqual(err, fakeApiError);
+        done();
+      });
+
+      userStream.emit('writing');
+      requestStream.end({error});
+    });
+
     it('re-emits response from the request stream', done => {
       // Stub the underlying _streamingRecognize method to just return
       // a bogus stream.