fix(Authorization): Speech to Text WebSockets recognize (#717)

This PR addresses the issues users were having with STT, specifically using streaming operations with IAM authentication.

Authenticating with an IAM API key requires requests to be made in order to get an access token. These request interrupt the stream and cause the streaming operation to fail. My solution for this is to give users an option (an option, but it's required for IAM and streams to work) to pre-authenticate and receive a `ready` flag that tells them they are good to proceed with the request. It is not the cleanest solution but I have not thought of a better way.

What this function does is make any IAM requests, if necessary. When the user makes their service request, the code will see that an access token is already stored and will launch the request without any delays that would interrupt the stream.

This line:
```js
const authHeader = { Authorization: 'Bearer ' + token };
this._options.headers = extend(authHeader, this._options.headers);
```

is for `createRecognizeStream`, which looks for an authorization header in `this._options.headers`.

An example of using this method:
```js
stt.preAuthenticate(function(ready) {
  if (!ready) {
    return;
  }
  stt.recognize(params, function(err, res) {
    if (err) {
      console.log(err);
    } else {
      console.log(JSON.stringify(res, null, 2));
    }
  });
});
```

We can definitely change the name too. Happy to take any suggestions.

TODO:
- [ ] Run tests
- [ ] Add a section about this in the README

Author: dpopp07

lib/base_service.ts

@@ -125,7 +125,6 @@ export class BaseService {
       options,
       _options
     );
-
     if (_options.iam_apikey || _options.iam_access_token) {
       this.tokenManager = new IamTokenManagerV1({
         iamApikey: _options.iam_apikey,
@@ -192,18 +191,40 @@ export class BaseService {
     }
   }
 
+  /**
+   * Guarantee that the next request you make will be IAM authenticated. This
+   * performs any requests necessary to get a valid IAM token so that if your
+   * next request involves a streaming operation, it will not be interrupted.
+   *
+   * @param {Function} callback - callback function to return flow of execution
+   *
+   * @returns {void}
+   */
+  protected preAuthenticate(callback): void {
+     if (Boolean(this.tokenManager)) {
+      return this.tokenManager.getToken((err, token) => {
+        if (err) {
+          callback(err);
+        }
+        callback(null);
+      });
+    } else {
+      callback(null);
+    }
+  }
+
   /**
    * Wrapper around `sendRequest` that determines whether or not IAM tokens
    * are being used to authenticate the request. If so, the token is 
    * retrieved by the token manager.
    *
    * @param {Object} parameters - service request options passed in by user
-   * @param {Function} callback - callback function to pass the reponse back to
+   * @param {Function} callback - callback function to pass the response back to
    * @returns {ReadableStream|undefined}
    */
   protected createRequest(parameters, callback) {
-     if (Boolean(this.tokenManager)) {
-      this.tokenManager.getToken((err, accessToken) => {
+    if (Boolean(this.tokenManager)) {
+      return this.tokenManager.getToken((err, accessToken) => {
         if (err) {
           return callback(err);
         }
@@ -215,6 +236,7 @@ export class BaseService {
       return sendRequest(parameters, callback);
     }
   }
+
   /**
    * @private
    * @param {UserOptions} options

lib/recognize-stream.ts

@@ -77,6 +77,7 @@ class RecognizeStream extends Duplex {
   private finished: boolean;
   private socket;
   private promise = require('./to-promise');
+  private authenticated: boolean;
 
 
 
@@ -110,6 +111,7 @@ class RecognizeStream extends Duplex {
    * @param {Number} [options.X-Watson-Learning-Opt-Out=false] - set to true to opt-out of allowing Watson to use this request to improve it's services
    * @param {Boolean} [options.smart_formatting=false] - formats numeric values such as dates, times, currency, etc.
    * @param {String} [options.customization_id] - Customization ID
+   * @param {IamTokenManagerV1} [options.token_manager] - Token manager for authenticating with IAM
    *
    * @constructor
    */
@@ -126,6 +128,8 @@ class RecognizeStream extends Duplex {
     this.listening = false;
     this.initialized = false;
     this.finished = false;
+    // is using iam, another authentication step is needed
+    this.authenticated = options.token_manager ? false : true;
     this.on('newListener', event => {
       if (!options.silent) {
         if (
@@ -392,39 +396,49 @@ class RecognizeStream extends Duplex {
     // so, the best we can do here is a no-op
   }
 
+
   _write(chunk, encoding, callback): void {
-    const self = this;
-    if (self.finished) {
-      // can't send any more data after the stop message (although this shouldn't happen normally...)
-      return;
-    }
-    if (!this.initialized) {
-      if (!this.options['content-type'] && !this.options.content_type) {
-        const ct = RecognizeStream.getContentType(chunk);
-        if (ct) {
-          this.options['content-type'] = ct;
-        } else {
-          const err = new Error(
-            'Unable to determine content-type from file header, please specify manually.'
-          );
-          err.name = RecognizeStream.ERROR_UNRECOGNIZED_FORMAT;
-          this.emit('error', err);
-          this.push(null);
-          return;
-        }
+    this.setAuthorizationHeaderToken(err => {
+      if (err) {
+        this.emit('error', err);
+        this.push(null);
+        return;
+      }
+      const self = this;
+      if (self.finished) {
+        // can't send any more data after the stop message (although this shouldn't happen normally...)
+        return;
       }
-      this.initialize();
 
-      this.once('open', () => {
+      if (!this.initialized) {
+        if (!this.options['content-type'] && !this.options.content_type) {
+          const ct = RecognizeStream.getContentType(chunk);
+          if (ct) {
+            this.options['content-type'] = ct;
+          } else {
+            const error = new Error(
+              'Unable to determine content-type from file header, please specify manually.'
+            );
+            error.name = RecognizeStream.ERROR_UNRECOGNIZED_FORMAT;
+            this.emit('error', error);
+            this.push(null);
+            return;
+          }
+        }
+        this.initialize();
+
+        this.once('open', () => {
+          self.sendData(chunk);
+          self.afterSend(callback);
+        });
+      } else {
         self.sendData(chunk);
-        self.afterSend(callback);
-      });
-    } else {
-      self.sendData(chunk);
-      this.afterSend(callback);
-    }
+        this.afterSend(callback);
+      }
+    })
   }
 
+
   finish(): void {
     // this is called both when the source stream finishes, and when .stop() is fired, but we only want to send the stop message once.
     if (this.finished) {
@@ -470,6 +484,31 @@ class RecognizeStream extends Duplex {
       }
     });
   }
+
+  /**
+   * This function retrieves an IAM access token and stores it in the
+   * request header before calling the callback function, which will
+   * execute the next iteration of `_write()`
+   *
+   *
+   * @private
+   * @param {Function} callback
+   */
+  setAuthorizationHeaderToken(callback) {
+    if (!this.authenticated) {
+      this.options.token_manager.getToken((err, token) => {
+        if (err) {
+          callback(err);
+        }
+        const authHeader = { authorization: 'Bearer ' + token };
+        this.options.headers = extend(authHeader, this.options.headers);
+        this.authenticated = true;
+        callback(null);
+      });
+    } else {
+      callback(null);
+    }
+  }
 }
 
 export = RecognizeStream;

lib/requestwrapper.ts

@@ -237,5 +237,6 @@ export function sendRequest(parameters, _callback) {
 
   // Compression support
   options.gzip = true;
+
   return request(options, formatErrorIfExists(_callback));
 }

speech-to-text/v1.ts

@@ -445,6 +445,18 @@ class SpeechToTextV1 extends GeneratedSpeechToTextV1 {
     params = params || {};
     params.url = this._options.url;
 
+    // if using iam, headers will not be a property on _options
+    // and the line `authorization: this._options.headers.Authorization`
+    // will crash the code
+    if (!this._options.headers) {
+      this._options.headers = {};
+    }
+
+    // if using iam, pass the token manager to the RecognizeStream object
+    if (this.tokenManager) {
+      params.token_manager = this.tokenManager;
+    }
+
     params.headers = extend(
       {
         'user-agent': pkg.name + '-nodejs-' + pkg.version,
@@ -502,20 +514,27 @@ class SpeechToTextV1 extends GeneratedSpeechToTextV1 {
       options: {
         method: 'POST',
         url: _url,
-        headers: {
-          'Content-Type': params.content_type
-        },
         json: true,
         qs: queryParams
       },
-      defaultOptions: this._options
-    };
-    return params.audio
-      .on('response', (response) => {
-        // Replace content-type
-        response.headers['content-type'] = params.content_type;
+      defaultOptions: extend(true, {}, this._options, {
+        headers: {
+          'Content-Type': params.content_type
+        }
       })
-      .pipe(this.createRequest(parameters, callback));
+    };
+
+    this.preAuthenticate((err) => {
+      if (err) {
+        return err;
+      }
+      return params.audio
+        .on('response', (response) => {
+          // Replace content-type
+          response.headers['content-type'] = params.content_type;
+        })
+        .pipe(this.createRequest(parameters, callback));
+    });
   }
 
   deleteCustomization(params, callback) {

test/integration/test.speech_to_text.js

@@ -28,8 +28,19 @@ describe('speech_to_text_integration', function() {
   });
 
   let speech_to_text;
+  let speech_to_text_rc;
+
   beforeEach(function() {
     speech_to_text = new watson.SpeechToTextV1(auth.speech_to_text);
+    speech_to_text_rc = new watson.SpeechToTextV1(auth.speech_to_text_rc);
+  });
+
+  it('recognize() (RC)', function(done) {
+    const params = {
+      audio: fs.createReadStream(path.join(__dirname, '../resources/weather.ogg')),
+      content_type: 'audio/ogg; codec=opus',
+    };
+    speech_to_text_rc.recognize(params, done);
   });
 
   it('recognize()', function(done) {
@@ -113,6 +124,90 @@ describe('speech_to_text_integration', function() {
     speech_to_text.getModels({}, done);
   });
 
+  describe('createRecognizeStream() (RC) (credentials from environment/VCAP)', () => {
+    let env;
+    beforeEach(function() {
+      env = process.env;
+      process.env = {};
+    });
+    afterEach(function() {
+      process.env = env;
+    });
+
+    it('transcribes audio over a websocket, credentials from environment', function(done) {
+      process.env.SPEECH_TO_TEXT_IAM_APIKEY = auth.speech_to_text_rc.iam_apikey;
+      process.env.SPEECH_TO_TEXT_URL = auth.speech_to_text_rc.url;
+      const speech_to_text_env = new watson.SpeechToTextV1({});
+      const recognizeStream = speech_to_text_env.createRecognizeStream();
+      recognizeStream.setEncoding('utf8');
+      fs
+        .createReadStream(path.join(__dirname, '../resources/weather.flac'))
+        .pipe(recognizeStream)
+        .on('error', done)
+        .pipe(
+          concat(function(transcription) {
+            assert.equal(typeof transcription, 'string', 'should return a string transcription');
+            assert.equal(
+              transcription.trim(),
+              'thunderstorms could produce large hail isolated tornadoes and heavy rain'
+            );
+            done();
+          })
+        );
+    });
+
+    it('transcribes audio over a websocket, credentials from VCAP_SERVICES', function(done) {
+      process.env.VCAP_SERVICES = JSON.stringify({
+        speech_to_text: [
+          {
+            credentials: {
+              iam_apikey: auth.speech_to_text_rc.iam_apikey,
+              url: auth.speech_to_text_rc.url,
+            },
+          },
+        ],
+      });
+      const speech_to_text_vcap = new watson.SpeechToTextV1({});
+      const recognizeStream = speech_to_text_vcap.createRecognizeStream();
+      recognizeStream.setEncoding('utf8');
+      fs
+        .createReadStream(path.join(__dirname, '../resources/weather.flac'))
+        .pipe(recognizeStream)
+        .on('error', done)
+        .pipe(
+          concat(function(transcription) {
+            assert.equal(typeof transcription, 'string', 'should return a string transcription');
+            assert.equal(
+              transcription.trim(),
+              'thunderstorms could produce large hail isolated tornadoes and heavy rain'
+            );
+            done();
+          })
+        );
+    });
+  });
+
+  describe('createRecognizeStream() (RC)', () => {
+    it('transcribes audio over a websocket', function(done) {
+      const recognizeStream = speech_to_text_rc.createRecognizeStream();
+      recognizeStream.setEncoding('utf8');
+      fs
+        .createReadStream(path.join(__dirname, '../resources/weather.flac'))
+        .pipe(recognizeStream)
+        .on('error', done)
+        .pipe(
+          concat(function(transcription) {
+            assert.equal(typeof transcription, 'string', 'should return a string transcription');
+            assert.equal(
+              transcription.trim(),
+              'thunderstorms could produce large hail isolated tornadoes and heavy rain'
+            );
+            done();
+          })
+        );
+    });
+  });
+
   describe('createRecognizeStream()', () => {
     it('transcribes audio over a websocket', function(done) {
       const recognizeStream = speech_to_text.createRecognizeStream();