Logcat works now, but still kind of wondering how the messages should arrive.

lib/roles/app.js

@@ -324,6 +324,9 @@ module.exports = function(options) {
       .on(wire.TransactionDoneMessage, function(channel, message) {
         socket.emit('tx.done', channel.toString(), message)
       })
+      .on(wire.DeviceLogcatEntryMessage, function(channel, message) {
+        socket.emit('logcat.entry', message)
+      })
       .handler()
 
     // Global messages
@@ -530,6 +533,26 @@ module.exports = function(options) {
               ])
             })
         })
+        .on('logcat.start', function(channel, responseChannel, data) {
+          joinChannel(responseChannel)
+          push.send([
+            channel
+          , wireutil.transaction(
+              responseChannel
+            , new wire.LogcatStartMessage(data)
+            )
+          ])
+        })
+        .on('logcat.stop', function(channel, responseChannel) {
+          joinChannel(responseChannel)
+          push.send([
+            channel
+          , wireutil.transaction(
+              responseChannel
+            , new wire.LogcatStopMessage()
+            )
+          ])
+        })
     })
     .finally(function() {
       // Clean up all listeners and subscriptions

lib/roles/device/plugins/group.js

@@ -19,13 +19,22 @@ module.exports = syrup.serial()
   .define(function(options, identity, input, router, push, sub, channels) {
     var log = logger.createLogger('device:plugins:group')
       , currentGroup = null
-      , emitter = new events.EventEmitter()
+      , plugin = new events.EventEmitter()
 
-    function joinGroup(newGroup, timeout) {
+    plugin.get = Promise.method(function() {
-      if (currentGroup) {
+      if (!currentGroup) {
-        return Promise.reject(new grouputil.AlreadyGroupedError())
+        throw new grouputil.NoGroupError()
       }
 
+      return currentGroup
+    })
+
+    plugin.join = function(newGroup, timeout) {
+      return plugin.get()
+        .then(function() {
+          throw new grouputil.AlreadyGroupedError()
+        })
+        .catch(grouputil.NoGroupError, function() {
           currentGroup = newGroup
 
           log.info('Now owned by "%s"', currentGroup.email)
@@ -42,48 +51,44 @@ module.exports = syrup.serial()
             ))
           ])
 
-      input.acquireWakeLock()
+          plugin.emit('join', currentGroup)
-      input.unlock()
 
-      emitter.emit('join', currentGroup)
+          return currentGroup
-
+        })
-      return Promise.resolve(currentGroup)
     }
 
-    function leaveGroup() {
+    plugin.leave = function() {
-      if (!currentGroup) {
+      return plugin.get()
-        return Promise.reject(new grouputil.NotGroupedError())
+        .then(function(group) {
-      }
+          log.info('No longer owned by "%s"', group.email)
+          log.info('Unsubscribing from group channel "%s"', group.group)
 
-      log.info('No longer owned by "%s"', currentGroup.email)
+          channels.unregister(group.group)
-      log.info('Unsubscribing from group channel "%s"', currentGroup.group)
+          sub.unsubscribe(group.group)
-
-      channels.unregister(currentGroup.group)
-      sub.unsubscribe(currentGroup.group)
 
           push.send([
             wireutil.global
           , wireutil.envelope(new wire.LeaveGroupMessage(
               options.serial
-        , currentGroup
+            , group
             ))
           ])
 
+          currentGroup = null
+          plugin.emit('leave', group)
+
+          return group
+        })
+    }
+
+    plugin.on('join', function() {
+      input.acquireWakeLock()
+      input.unlock()
+    })
+
+    plugin.on('leave', function() {
       input.releaseWakeLock()
       input.lock()
-
-      var oldGroup = currentGroup
-      currentGroup = null
-
-      emitter.emit('leave', oldGroup)
-
-      return Promise.resolve(oldGroup)
-    }
-
-    channels.on('timeout', function(channel) {
-      if (currentGroup && channel === currentGroup.group) {
-        leaveGroup()
-      }
     })
 
     router
@@ -91,7 +96,7 @@ module.exports = syrup.serial()
         var reply = wireutil.reply(options.serial)
         grouputil.match(identity, message.requirements)
           .then(function() {
-            return joinGroup(message.owner)
+            return plugin.join(message.owner, message.timeout)
           })
           .then(function() {
             push.send([
@@ -116,7 +121,7 @@ module.exports = syrup.serial()
         var reply = wireutil.reply(options.serial)
         grouputil.match(identity, message.requirements)
           .then(function() {
-            return leaveGroup()
+            return plugin.leave()
           })
           .then(function() {
             push.send([
@@ -124,7 +129,7 @@ module.exports = syrup.serial()
             , reply.okay()
             ])
           })
-          .catch(grouputil.NotGroupedError, function(err) {
+          .catch(grouputil.NoGroupError, function(err) {
             push.send([
               channel
             , reply.fail(err.message)
@@ -132,15 +137,19 @@ module.exports = syrup.serial()
           })
       })
 
-    lifecycle.observe(function() {
+    channels.on('timeout', function(channel) {
-      if (currentGroup) {
+      if (currentGroup && channel === currentGroup.group) {
-        leaveGroup()
+        plugin.leave()
-        return Promise.delay(500)
-      }
-      else {
-        return true
       }
     })
 
-    return emitter
+    lifecycle.observe(function() {
+      return plugin.leave()
+        .delay(500) // Make sure that the message gets sent
+        .catch(grouputil.NoGroupError, function() {
+          return true
+        })
+    })
+
+    return plugin
   })

lib/roles/device/plugins/logcat.js

@@ -1,4 +1,5 @@
 var syrup = require('syrup')
+var Promise = require('bluebird')
 
 var logger = require('../../../util/logger')
 var wire = require('../../../wire')
@@ -10,29 +11,29 @@ module.exports = syrup.serial()
   .dependency(require('../support/router'))
   .dependency(require('../support/push'))
   .dependency(require('./group'))
-  .define(function(options, adb, router, push, owner) {
+  .define(function(options, adb, router, push, group) {
     var log = logger.createLogger('device:plugins:logcat')
+      , plugin = Object.create(null)
+      , activeLogcat = null
 
-    function openService() {
+    plugin.start = function(filters) {
-      log.info('Launching logcat service')
+      return group.get()
-      return adb.openLogcat(options.serial)
+        .then(function(group) {
+          return plugin.stop()
+            .then(function() {
+              log.info('Starting logcat')
+              return adb.openLogcat(options.serial, {
+                clear: true
+              })
+            })
             .timeout(10000)
             .then(function(logcat) {
-          return lifecycle.share('Logcat', logcat)
+              activeLogcat = logcat
-        })
-    }
-
-    return openService()
-      .then(function(logcat) {
-        function reset() {
-          logcat
-            .resetFilters()
-            .excludeAll()
-        }
 
               function entryListener(entry) {
+                console.log('sending', entry)
                 push.send([
-            owner.group
+                  group.group
                 , wireutil.envelope(new wire.DeviceLogcatEntryMessage(
                     options.serial
                   , entry.date.getTime() / 1000
@@ -45,17 +46,95 @@ module.exports = syrup.serial()
                 ])
               }
 
-        reset()
               logcat.on('entry', entryListener)
 
+              return plugin.reset(filters)
+            })
+        })
+    }
+
+    plugin.stop = Promise.method(function() {
+      if (plugin.isRunning()) {
+        log.info('Stopping logcat')
+        activeLogcat.end()
+        activeLogcat = null
+      }
+    })
+
+    plugin.reset = Promise.method(function(filters) {
+      if (plugin.isRunning()) {
+        activeLogcat
+          .resetFilters()
+          .excludeAll()
+
+        filters.forEach(function(filter) {
+          activeLogcat.include(filter.tag, filter.priority)
+        })
+      }
+      else {
+        throw new Error('Logcat is not running')
+      }
+    })
+
+    plugin.isRunning = function() {
+      return !!activeLogcat
+    }
+
+    lifecycle.observe(plugin.stop)
+    group.on('leave', plugin.stop)
+
     router
+      .on(wire.LogcatStartMessage, function(channel, message) {
+        var reply = wireutil.reply(options.serial)
+        plugin.start(message.filters)
+          .then(function() {
+            push.send([
+              channel
+            , reply.okay('success')
+            ])
+          })
+          .catch(function(err) {
+            log.error('Unable to open logcat', err.stack)
+            push.send([
+              channel
+            , reply.fail('fail')
+            ])
+          })
+      })
       .on(wire.LogcatApplyFiltersMessage, function(channel, message) {
-            reset()
+        var reply = wireutil.reply(options.serial)
-            message.filters.forEach(function(filter) {
+        plugin.reset(message.filters)
-              logcat.include(filter.tag, filter.priority)
+          .then(function() {
+            push.send([
+              channel
+            , reply.okay('success')
+            ])
+          })
+          .catch(function(err) {
+            log.error('Failed to apply logcat filters', err.stack)
+            push.send([
+              channel
+            , reply.fail('fail')
+            ])
+          })
+      })
+      .on(wire.LogcatStopMessage, function(channel) {
+        var reply = wireutil.reply(options.serial)
+        plugin.stop()
+          .then(function() {
+            push.send([
+              channel
+            , reply.okay('success')
+            ])
+          })
+          .catch(function(err) {
+            log.error('Failed to stop logcat', err.stack)
+            push.send([
+              channel
+            , reply.fail('fail')
+            ])
           })
       })
 
-        return logcat
+    return plugin
-      })
   })

lib/roles/processor.js

@@ -91,6 +91,9 @@ module.exports = function(options) {
     .on(wire.TransactionDoneMessage, function(channel, message, data) {
       appDealer.send([channel, data])
     })
+    .on(wire.DeviceLogcatEntryMessage, function(channel, message, data) {
+      appDealer.send([channel, data])
+    })
     .handler())
 
   lifecycle.observe(function() {

lib/util/grouputil.js

@@ -28,16 +28,16 @@ util.inherits(AlreadyGroupedError, Error)
 
 module.exports.AlreadyGroupedError = AlreadyGroupedError
 
-function NotGroupedError() {
+function NoGroupError() {
   Error.call(this)
-  this.name = 'NotGroupedError'
+  this.name = 'NoGroupError'
   this.message = 'Not a member of any group'
-  Error.captureStackTrace(this, NotGroupedError)
+  Error.captureStackTrace(this, NoGroupError)
 }
 
-util.inherits(NotGroupedError, Error)
+util.inherits(NoGroupError, Error)
 
-module.exports.NotGroupedError = NotGroupedError
+module.exports.NoGroupError = NoGroupError
 
 module.exports.match = Promise.method(function(capabilities, requirements) {
   return requirements.every(function(req) {

lib/wire/wire.proto

@@ -39,6 +39,8 @@ enum MessageType {
   ForwardTestMessage         = 36;
   ForwardCreateMessage       = 37;
   ForwardRemoveMessage       = 38;
+  LogcatStartMessage         = 39;
+  LogcatStopMessage          = 40;
 }
 
 message Envelope {
@@ -316,6 +318,13 @@ message LogcatFilter {
   required uint32 priority = 2;
 }
 
+message LogcatStartMessage {
+  repeated LogcatFilter filters = 1;
+}
+
+message LogcatStopMessage {
+}
+
 message LogcatApplyFiltersMessage {
   repeated LogcatFilter filters = 1;
 }

res/app/components/stf/control/control-service.js

@@ -187,6 +187,16 @@ module.exports = function ControlServiceFactory(
         devicePort: forward.devicePort
       })
     }
+
+    this.startLogcat = function(filters) {
+      return sendTwoWay('logcat.start', {
+        filters: filters
+      })
+    }
+
+    this.stopLogcat = function() {
+      return sendTwoWay('logcat.stop')
+    }
   }
 
   controlService.create = function(target, channel) {