Keep better track of vital streams. Add monkey.

lib/roles/device.js

@@ -11,6 +11,7 @@ var wireutil = require('../util/wireutil')(wire)
 var devutil = require('../util/devutil')
 var pathutil = require('../util/pathutil')
 var promiseutil = require('../util/promiseutil')
+var Vitals = require('../util/vitals')
 var ChannelManager = require('../wire/channelmanager')
 
 module.exports = function(options) {
@@ -19,6 +20,17 @@ module.exports = function(options) {
   var vendor = Object.create(null)
   var solo = wireutil.makePrivateChannel()
   var channels = new ChannelManager()
+  var vitals = new Vitals()
+  var ports = {
+    http: 2870
+  , input: 2820
+  , stats: 2830
+  , forward: 2810
+  }
+  var services = {
+    input: null
+  , monkey: null
+  }
 
   // Show serial number in logs
   logger.setGlobalIdentifier(options.serial)
@@ -47,6 +59,18 @@ module.exports = function(options) {
     push.send([channel, wireutil.makeLeaveGroupMessage(options.serial)])
   })
 
+  // Closure of vital functionality
+  vitals.on('end', function(name) {
+    log.fatal(util.format('Vital utility "%s" has ended', name))
+    selfDestruct()
+  })
+
+  // Error in vital utility
+  vitals.on('error', function(name, err) {
+    log.fatal(util.format('Vital utility "%s" had an error', er))
+    selfDestruct()
+  })
+
   promiseutil.periodicNotify(adb.waitBootCompleteAsync(options.serial), 1000)
     .progressed(function() {
       log.info('Waiting for boot to complete')
@@ -90,21 +114,98 @@ module.exports = function(options) {
       ])
     })
     .then(function() {
+      log.info('Launching HTTP API')
       return devutil.ensureUnusedPort(adb, options.serial, 2870)
         .then(function(port) {
+          var log = logger.createLogger('device:remote:http')
           return adb.shellAsync(options.serial, [
               vendor.bin.dest
             , '--lib', vendor.lib.dest
             , '--listen-http', port
             ])
             .then(function(out) {
+              vitals.register('device:remote:http:shell', out)
               out.pipe(require('split')())
                 .on('data', function(chunk) {
-                  log.info('remote: "%s"', chunk)
+                  log.info(chunk)
                 })
-                .on('end', function() {
-                  log.fatal('remote: Connection closed')
-                  selfDestruct()
+            })
+        })
+    })
+    .then(function() {
+      log.info('Launching monkey service')
+      return devutil.ensureUnusedPort(adb, options.serial, 1080)
+        .then(function(port) {
+          var log = logger.createLogger('device:remote:monkey')
+          return adb.shellAsync(options.serial, [
+              'monkey'
+            , '--port', port
+            ])
+            .then(function(out) {
+              vitals.register('device:remote:monkey:shell', out)
+              out.pipe(require('split')())
+                .on('data', function(chunk) {
+                  log.info(chunk)
+                })
+              return port
+            })
+        })
+        .then(function(port) {
+          return devutil.waitForPort(adb, options.serial, port)
+        })
+        .then(function(port) {
+          return adb.openMonkeyAsync(options.serial, port)
+        })
+        .then(function(monkey) {
+          services.monkey =
+            vitals.register('device:remote:monkey:monkey', monkey)
+        })
+    })
+    .then(function() {
+      log.info('Launching input service')
+      return devutil.ensureUnusedPort(adb, options.serial, 2820)
+        .then(function(port) {
+          var log = logger.createLogger('device:remote:input')
+          return adb.shellAsync(options.serial, [
+              vendor.bin.dest
+            , '--lib', vendor.lib.dest
+            , '--listen-input', port
+            ])
+            .then(function(out) {
+              vitals.register('device:remote:input:shell', out)
+              out.pipe(require('split')())
+                .on('data', function(chunk) {
+                  log.info(chunk)
+                })
+              return port
+            })
+        })
+        .then(function(port) {
+          return devutil.waitForPort(adb, options.serial, port)
+        })
+        .then(function(port) {
+          return adb.openMonkeyAsync(options.serial, port)
+        })
+        .then(function(monkey) {
+          services.input =
+            vitals.register('device:remote:input:monkey', monkey)
+        })
+    })
+    .then(function() {
+      log.info('Launching stats service')
+      return devutil.ensureUnusedPort(adb, options.serial, 2830)
+        .then(function(port) {
+          var log = logger.createLogger('device:remote:stats')
+          return adb.shellAsync(options.serial, [
+              vendor.bin.dest
+            , '--lib', vendor.lib.dest
+            , '--listen-stats', port
+            ])
+            .then(function(out) {
+              vitals.register('device:remote:stats:shell', out)
+              out.pipe(require('split')())
+                .on('data', function(chunk) {
+                  log.info(chunk)
                 })
             })
         })

lib/util/devutil.js

@@ -40,6 +40,23 @@ devutil.ensureUnusedPort = function(adb, serial, port) {
     })
 }
 
+devutil.waitForPort = function(adb, serial, port) {
+  function closedError(err) {
+    return err.message === 'closed'
+  }
+  return adb.openTcpAsync(serial, port)
+    .then(function(conn) {
+      conn.end()
+      return port
+    })
+    .catch(closedError, function(err) {
+      return Promise.delay(100)
+        .then(function() {
+          return devutil.waitForPort(adb, serial, port)
+        })
+    })
+}
+
 devutil.killProcsByComm = function(adb, serial, comm, bin) {
   return adb.shellAsync(serial, ['ps', comm])
     .then(function(out) {

lib/util/vitals.js

@@ -0,0 +1,24 @@
+var events = require('events')
+var util = require('util')
+
+function Vitals() {
+  events.EventEmitter.call(this)
+}
+
+util.inherits(Vitals, events.EventEmitter)
+
+Vitals.prototype.register = function(name, stream) {
+  var that = this
+
+  stream.on('end', function() {
+    that.emit('end', name)
+  })
+
+  stream.on('error', function(err) {
+    that.emit('error', name, err)
+  })
+
+  return stream
+}
+
+module.exports = Vitals