add mjpeg stream helper class

index.js

@@ -10,8 +10,9 @@ import * as logger from './lib/logging';
 import * as process from './lib/process';
 import * as zip from './lib/zip';
 import * as imageUtil from './lib/image-util';
+import * as mjpeg from './lib/mjpeg';
 
 // can't add to other exports `as default`
 // until JSHint figures out how to parse that pattern
 export default { tempDir, system, util, fs, cancellableDelay, plist, mkdirp,
-                 logger, process, zip, imageUtil, net };
+                 logger, process, zip, imageUtil, net, mjpeg };

lib/mjpeg.js

@@ -0,0 +1,184 @@
+import _ from 'lodash';
+import request from 'request';
+import log from './logger';
+import http from 'http';
+import B from 'bluebird';
+import Jimp from 'jimp';
+import MJpegConsumer from 'mjpeg-consumer';
+import mJpegServer from 'mjpeg-server';
+import { Writable } from 'stream';
+
+const TEST_IMG_JPG = '/9j/4QAYRXhpZgAASUkqAAgAAAAAAAAAAAAAAP/sABFEdWNreQABAAQAAAAeAAD/4QOBaHR0cDovL25zLmFkb2JlLmNvbS94YXAvMS4wLwA8P3hwYWNrZXQgYmVnaW49Iu+7vyIgaWQ9Ilc1TTBNcENlaGlIenJlU3pOVGN6a2M5ZCI/PiA8eDp4bXBtZXRhIHhtbG5zOng9ImFkb2JlOm5zOm1ldGEvIiB4OnhtcHRrPSJBZG9iZSBYTVAgQ29yZSA1LjYtYzE0MCA3OS4xNjA0NTEsIDIwMTcvMDUvMDYtMDE6MDg6MjEgICAgICAgICI+IDxyZGY6UkRGIHhtbG5zOnJkZj0iaHR0cDovL3d3dy53My5vcmcvMTk5OS8wMi8yMi1yZGYtc3ludGF4LW5zIyI+IDxyZGY6RGVzY3JpcHRpb24gcmRmOmFib3V0PSIiIHhtbG5zOnhtcE1NPSJodHRwOi8vbnMuYWRvYmUuY29tL3hhcC8xLjAvbW0vIiB4bWxuczpzdFJlZj0iaHR0cDovL25zLmFkb2JlLmNvbS94YXAvMS4wL3NUeXBlL1Jlc291cmNlUmVmIyIgeG1sbnM6eG1wPSJodHRwOi8vbnMuYWRvYmUuY29tL3hhcC8xLjAvIiB4bXBNTTpPcmlnaW5hbERvY3VtZW50SUQ9InhtcC5kaWQ6NGY5ODc1OTctZGE2My00Y2M0LTkzNDMtNGYyNjgzMGUwNjk3IiB4bXBNTTpEb2N1bWVudElEPSJ4bXAuZGlkOjlDMzI3QkY0N0Q3NTExRThCMTlDOTVDMDc2RDE5MDY5IiB4bXBNTTpJbnN0YW5jZUlEPSJ4bXAuaWlkOjlDMzI3QkYzN0Q3NTExRThCMTlDOTVDMDc2RDE5MDY5IiB4bXA6Q3JlYXRvclRvb2w9IkFkb2JlIFBob3Rvc2hvcCBDQyAyMDE4IChNYWNpbnRvc2gpIj4gPHhtcE1NOkRlcml2ZWRGcm9tIHN0UmVmOmluc3RhbmNlSUQ9InhtcC5paWQ6NGY5ODc1OTctZGE2My00Y2M0LTkzNDMtNGYyNjgzMGUwNjk3IiBzdFJlZjpkb2N1bWVudElEPSJ4bXAuZGlkOjRmOTg3NTk3LWRhNjMtNGNjNC05MzQzLTRmMjY4MzBlMDY5NyIvPiA8L3JkZjpEZXNjcmlwdGlvbj4gPC9yZGY6UkRGPiA8L3g6eG1wbWV0YT4gPD94cGFja2V0IGVuZD0iciI/Pv/uAA5BZG9iZQBkwAAAAAH/2wCEABALCwsMCxAMDBAXDw0PFxsUEBAUGx8XFxcXFx8eFxoaGhoXHh4jJSclIx4vLzMzLy9AQEBAQEBAQEBAQEBAQEABEQ8PERMRFRISFRQRFBEUGhQWFhQaJhoaHBoaJjAjHh4eHiMwKy4nJycuKzU1MDA1NUBAP0BAQEBAQEBAQEBAQP/AABEIACAAIAMBIgACEQEDEQH/xABgAAEAAwEAAAAAAAAAAAAAAAAABAUHCAEBAAAAAAAAAAAAAAAAAAAAABAAAQMCAgsAAAAAAAAAAAAAAAECBBEDEgYhMRODo7PTVAUWNhEBAAAAAAAAAAAAAAAAAAAAAP/aAAwDAQACEQMRAD8Az8AAdAAAAAAI8+fE8dEuTZtzZR7VMb6OdTE5GJoYirrUp/e8qd9wb3TGe/lJ2551sx8D/9k=';
+
+// amount of time to wait for the first image in the stream
+const MJPEG_SERVER_TIMEOUT_MS = 10000;
+
+/** Class which stores the last bit of data streamed into it */
+class MJpegStream extends Writable {
+
+  /**
+   * Create an MJpegStream
+   * @param {string} mJpegUrl - URL of MJPEG-over-HTTP stream
+   * @param {function} [errorHandler=noop] - additional function that will be
+   * called in the case of any errors.
+   * @param {object} [options={}] - Options to pass to the Writable constructor
+   */
+  constructor (mJpegUrl, errorHandler = _.noop, options = {}) {
+    super(options);
+    this.errorHandler = errorHandler;
+    this.url = mJpegUrl;
+    this.clear();
+  }
+
+  /**
+   * Get the base64-encoded version of the JPEG
+   * @returns {string}
+   */
+  get lastChunkBase64 () {
+    return _.isBuffer(this.lastChunk) ?
+      this.lastChunk.toString('base64') :
+      null;
+  }
+
+  /**
+   * Get the PNG version of the JPEG buffer
+   * @returns {Buffer} PNG image data
+   */
+  async lastChunkPNG () {
+    if (!_.isBuffer(this.lastChunk)) {
+      return null;
+    }
+
+    const jpg = await new B((res, rej) => {
+      new Jimp(this.lastChunk, (err, img) => {
+        if (err) {
+          return rej(err);
+        }
+        res(img);
+      });
+    });
+
+    return await B.promisify(jpg.getBuffer, {context: jpg})(Jimp.MIME_PNG);
+  }
+
+  /**
+   * Get the base64-encoded version of the PNG
+   * @returns {string}
+   */
+  async lastChunkPNGBase64 () {
+    const png = await this.lastChunkPNG();
+
+    if (!png) {
+      return null;
+    }
+
+    return png.toString('base64');
+  }
+
+  /**
+   * Reset internal state
+   */
+  clear () {
+    this.registerStartSuccess = null;
+    this.registerStartFailure = null;
+    this.request = null;
+    this.consumer = null;
+    this.lastChunk = null;
+    this.updateCount = 0;
+  }
+
+  /**
+   * Start reading the MJpeg stream and storing the last image
+   */
+  async start (serverTimeout = MJPEG_SERVER_TIMEOUT_MS) {
+    // ensure we're not started already
+    this.stop();
+
+    this.consumer = new MJpegConsumer();
+
+    // use the deferred pattern so we can wait for the start of the stream
+    // based on what comes in from an external pipe
+    const startPromise = new B((res, rej) => {
+      this.registerStartSuccess = res;
+      this.registerStartFailure = rej;
+    })
+    // start a timeout so that if the server does not return data, we don't
+    // block forever.
+      .timeout(serverTimeout,
+        `Waited ${serverTimeout}ms but the MJPEG server never sent any images`);
+
+    const onErr = (err) => {
+      log.error(`Error getting MJpeg screenshot chunk: ${err}`);
+      this.errorHandler(err);
+      if (this.registerStartFailure) {
+        this.registerStartFailure(err);
+      }
+    };
+
+    this.request = request(this.url);
+
+    this.request
+      .on('error', onErr)   // ensure we do something with errors
+      .pipe(this.consumer)  // allow chunking and transforming of jpeg data
+      .pipe(this);          // send the actual jpegs to ourself
+
+    await startPromise;
+  }
+
+  /**
+   * Stop reading the MJpeg stream. Ensure we disconnect all the pipes and stop
+   * the HTTP request itself. Then reset the state.
+   */
+  stop () {
+    if (!this.consumer) {
+      return;
+    }
+
+    this.consumer.unpipe();
+    this.request.end();
+    this.clear();
+  }
+
+  /**
+   * Override the Writable write() method in order to save the last image and
+   * log the number of images we have received
+   * @override
+   * @param {Buffer} data - binary data streamed from the MJpeg consumer
+   */
+  write (data) {
+    this.lastChunk = data;
+    this.updateCount++;
+
+    if (this.registerStartSuccess) {
+      this.registerStartSuccess();
+      this.registerStartSuccess = null;
+    }
+  }
+}
+
+/**
+ * Start an mjpeg server for the purpose of testing, which just sends the same
+ * image over and over. Caller is responsible for closing the server.
+ * @param {int} port - port the server should listen on
+ * @param {int} [intMs] - how often the server should push an image
+ * @param {int} [times] - how many times the server should push an image before
+ * it closes the connection
+ * @returns {http.Server}
+ */
+function initMJpegServer (port, intMs = 300, times = 20) {
+  const server = http.createServer(async function (req, res) {
+    const mJpegReqHandler = mJpegServer.createReqHandler(req, res);
+    const jpg = Buffer.from(TEST_IMG_JPG, 'base64');
+
+    // just send the same jpeg over and over
+    for (let i = 0; i < times; i++) {
+      await B.delay(intMs);
+      mJpegReqHandler._write(jpg, null, _.noop);
+    }
+    mJpegReqHandler.close();
+  }).listen(port);
+
+  return server;
+}
+
+export { MJpegStream, initMJpegServer, TEST_IMG_JPG };

package.json

@@ -30,9 +30,12 @@
     "bplist-parser": "^0.1.0",
     "extract-zip": "^1.6.0",
     "glob": "^7.1.2",
+    "jimp": "^0.2.28",
     "jsftp": "^2.1.2",
     "lodash": "^4.2.1",
     "md5-file": "^4.0.0",
+    "mjpeg-consumer": "^1.1.0",
+    "mjpeg-server": "^0.3.0",
     "mkdirp": "^0.5.1",
     "mv": "^2.1.1",
     "ncp": "^2.0.0",

test/mjpeg-specs.js

@@ -0,0 +1,69 @@
+import _ from 'lodash';
+import { mjpeg } from '..';
+import chai from 'chai';
+import chaiAsPromised from 'chai-as-promised';
+import B from 'bluebird';
+
+const {MJpegStream, initMJpegServer} = mjpeg;
+
+const should = chai.should();
+chai.use(chaiAsPromised);
+
+const MJPEG_SERVER_PORT = 8589;
+const MJPEG_SERVER_URL = `http://localhost:${MJPEG_SERVER_PORT}`;
+
+describe('MJpeg Stream (e2e)', function () {
+
+  let mJpegServer, stream;
+
+  before(function () {
+    mJpegServer = initMJpegServer(MJPEG_SERVER_PORT);
+  });
+
+  after(function () {
+    mJpegServer.close();
+    stream.stop(); // ensure streams are always stopped
+  });
+
+  it('should update mjpeg stream based on new data from mjpeg server', async function () {
+    stream = new MJpegStream(MJPEG_SERVER_URL, _.noop);
+    should.not.exist(stream.lastChunk);
+    await stream.start();
+    should.exist(stream.lastChunk);
+    stream.updateCount.should.eql(1);
+
+    await B.delay(1000); // let the stream update a bit
+    stream.updateCount.should.be.above(1);
+
+    // verify jpeg type and byte length of fixture image
+    const startBytes = Buffer.from([0xff, 0xd8]);
+    const endBytes = Buffer.from([0xff, 0xd9]);
+    const startPos = stream.lastChunk.indexOf(startBytes);
+    const endPos = stream.lastChunk.indexOf(endBytes);
+    startPos.should.eql(0);   // proves we have a jpeg
+    endPos.should.eql(1278); // proves we have a jpeg of the right size
+
+    // verify we can get the base64 version too
+    const b64 = stream.lastChunkBase64;
+    b64.should.eql(mjpeg.TEST_IMG_JPG);
+
+    // verify we can get the PNG version too
+    const png = await stream.lastChunkPNGBase64();
+    png.should.be.a('string');
+    png.indexOf('iVBOR').should.eql(0);
+    png.length.should.be.above(400);
+
+
+    // now stop the stream and wait some more then assert no new data has come in
+    stream.stop();
+    await B.delay(1000);
+    should.not.exist(stream.lastChunk);
+    stream.updateCount.should.eql(0);
+  });
+
+  it('should error out if the server does not send any images before a timeout', async function () {
+    stream = new MJpegStream(MJPEG_SERVER_URL, _.noop);
+    await stream.start(0).should.eventually.be.rejectedWith(/never sent/);
+  });
+
+});