Reaper decides if a device is present or not. Allows devices to "come…

… back to life" if they start beating again.

lib/db/api.js

@@ -120,14 +120,15 @@ dbapi.saveDeviceLog = function(serial, entry) {
     }))
 }
 
-dbapi.saveDevice = function(serial, device) {
+dbapi.saveDeviceInitialState = function(serial, device) {
   var data = {
-    present: true
+    present: false
+  , presenceChangedAt: r.now()
   , provider: device.provider
   , owner: null
   , status: device.status
-  , ready: false
   , statusChangedAt: r.now()
+  , ready: false
   , reverseForwards: []
   }
   return db.run(r.table('devices').get(serial).update(data))
@@ -160,10 +161,17 @@ dbapi.unsetDeviceOwner = function(serial) {
   }))
 }
 
+dbapi.setDevicePresent = function(serial) {
+  return db.run(r.table('devices').get(serial).update({
+    present: true
+  , presenceChangedAt: r.now()
+  }))
+}
+
 dbapi.setDeviceAbsent = function(serial) {
   return db.run(r.table('devices').get(serial).update({
     present: false
-  , ready: false
+  , presenceChangedAt: r.now()
   }))
 }
 

lib/units/processor/index.js

@@ -54,8 +54,8 @@ module.exports = function(options) {
 
   devDealer.on('message', wirerouter()
     // Initial device message
-    .on(wire.DevicePresentMessage, function(channel, message, data) {
-      dbapi.saveDevice(message.serial, message)
+    .on(wire.DeviceIntroductionMessage, function(channel, message, data) {
+      dbapi.saveDeviceInitialState(message.serial, message)
         .then(function() {
           devDealer.send([
             message.provider.channel
@@ -67,6 +67,10 @@ module.exports = function(options) {
         })
     })
     // Workerless messages
+    .on(wire.DevicePresentMessage, function(channel, message, data) {
+      dbapi.setDevicePresent(message.serial)
+      appDealer.send([channel, data])
+    })
     .on(wire.DeviceAbsentMessage, function(channel, message, data) {
       dbapi.setDeviceAbsent(message.serial)
       appDealer.send([channel, data])

lib/units/provider/index.js

@@ -151,7 +151,7 @@ module.exports = function(options) {
         // Tell others we found a device
         push.send([
           wireutil.global
-        , wireutil.envelope(new wire.DevicePresentMessage(
+        , wireutil.envelope(new wire.DeviceIntroductionMessage(
             device.id
           , wireutil.toDeviceStatus(device.type)
           , new wire.ProviderMessage(

lib/units/reaper/index.js

@@ -56,7 +56,17 @@ module.exports = function(options) {
     lifecycle.fatal()
   })
 
-  ttlset.on('timeout', function(serial) {
+  ttlset.on('insert', function(serial) {
+    log.info('Device "%s" is present', serial)
+    push.send([
+      wireutil.global
+    , wireutil.envelope(new wire.DevicePresentMessage(
+        serial
+      ))
+    ])
+  })
+
+  ttlset.on('drop', function(serial) {
     log.info('Reaping device "%s" due to heartbeat timeout', serial)
     push.send([
       wireutil.global
@@ -73,22 +83,22 @@ module.exports = function(options) {
           .then(function(list) {
             var now = Date.now()
             list.forEach(function(device) {
-              ttlset.bump(device.serial, now)
+              ttlset.bump(device.serial, now, TtlSet.SILENT)
             })
           })
       })
   }
 
   function listenToChanges() {
     sub.on('message', wirerouter()
-      .on(wire.DevicePresentMessage, function(channel, message) {
+      .on(wire.DeviceIntroductionMessage, function(channel, message) {
         ttlset.bump(message.serial, Date.now())
       })
       .on(wire.DeviceHeartbeatMessage, function(channel, message) {
         ttlset.bump(message.serial, Date.now())
       })
       .on(wire.DeviceAbsentMessage, function(channel, message) {
-        ttlset.drop(message.serial)
+        ttlset.drop(message.serial, TtlSet.SILENT)
       })
       .handler())
   }

lib/units/websocket/index.js

@@ -120,26 +120,35 @@ module.exports = function(options) {
       .on(wire.DeviceLogMessage, function(channel, message) {
         socket.emit('device.log', message)
       })
-      .on(wire.DevicePresentMessage, function(channel, message) {
+      .on(wire.DeviceIntroductionMessage, function(channel, message) {
         socket.emit('device.add', {
           important: true
         , data: {
             serial: message.serial
+          , present: false
           , provider: message.provider
-          , present: true
           , owner: null
+          , status: message.status
+          , ready: false
           , reverseForwards: []
           }
         })
       })
+      .on(wire.DevicePresentMessage, function(channel, message) {
+        socket.emit('device.change', {
+          important: true
+        , data: {
+            serial: message.serial
+          , present: true
+          }
+        })
+      })
       .on(wire.DeviceAbsentMessage, function(channel, message) {
         socket.emit('device.remove', {
           important: true
         , data: {
             serial: message.serial
           , present: false
-          , ready: false
-          , using: false
           , likelyLeaveReason: 'device_absent'
           }
         })

lib/util/ttlset.js

@@ -20,8 +20,10 @@ function TtlSet(ttl) {
 
 util.inherits(TtlSet, EventEmitter)
 
-TtlSet.prototype.bump = function(value, time) {
-  var item = this._remove(this.mapping[value]) || this._create(value)
+TtlSet.SILENT = 1
+
+TtlSet.prototype.bump = function(value, time, flags) {
+  var item = this._remove(this.mapping[value]) || this._create(value, flags)
 
   item.time = time || Date.now()
   item.prev = this.tail
@@ -37,8 +39,8 @@ TtlSet.prototype.bump = function(value, time) {
   }
 }
 
-TtlSet.prototype.drop = function(value) {
-  this._drop(this.mapping[value])
+TtlSet.prototype.drop = function(value, flags) {
+  this._drop(this.mapping[value], flags)
 }
 
 TtlSet.prototype.stop = function() {
@@ -59,8 +61,7 @@ TtlSet.prototype._check = function() {
   var item
   while ((item = this.head)) {
     if (now - item.time > this.ttl) {
-      this._drop(item)
-      this.emit('timeout', item.value)
+      this._drop(item, 0)
     }
     else {
       break
@@ -70,16 +71,27 @@ TtlSet.prototype._check = function() {
   this._scheduleCheck()
 }
 
-TtlSet.prototype._create = function(value) {
+TtlSet.prototype._create = function(value, flags) {
   var item = new TtlItem(value)
+
   this.mapping[value] = item
+
+  if ((flags & TtlSet.SILENT) !== TtlSet.SILENT) {
+    this.emit('insert', value)
+  }
+
   return item
 }
 
-TtlSet.prototype._drop = function(item) {
+TtlSet.prototype._drop = function(item, flags) {
   if (item) {
     this._remove(item)
+
     delete this.mapping[item.value]
+
+    if ((flags & TtlSet.SILENT) !== TtlSet.SILENT) {
+      this.emit('drop', item.value)
+    }
   }
 }
 

lib/wire/wire.proto

@@ -2,6 +2,7 @@
 
 enum MessageType {
   CopyMessage                = 33;
+  DeviceIntroductionMessage  = 74;
   DeviceAbsentMessage        = 1;
   DeviceIdentityMessage      = 2;
   DeviceLogcatEntryMessage   = 3;
@@ -117,7 +118,7 @@ message DeviceHeartbeatMessage {
   required string serial = 1;
 }
 
-message DevicePresentMessage {
+message DeviceIntroductionMessage {
   required string serial = 1;
   required DeviceStatus status = 2;
   required ProviderMessage provider = 3;
@@ -127,6 +128,10 @@ message DeviceRegisteredMessage {
   required string serial = 1;
 }
 
+message DevicePresentMessage {
+  required string serial = 1;
+}
+
 message DeviceAbsentMessage {
   required string serial = 1;
 }

test/util/ttlset.js

@@ -7,11 +7,11 @@ var TtlSet = require('../../lib/util/ttlset')
 
 describe('TtlSet', function() {
 
-  it.only('should timeout old entries', function(done) {
+  it('should emit "drop" for entries with expired TTL', function(done) {
     var ttlset = new TtlSet(50)
 
     var spy = sinon.spy()
-    ttlset.on('timeout', spy)
+    ttlset.on('drop', spy)
 
     ttlset.bump(1, Date.now())
     ttlset.bump(2, Date.now() + 100)
@@ -31,6 +31,42 @@ describe('TtlSet', function() {
 
   describe('bump', function() {
 
+    it('should emit "insert" for new entries', function(done) {
+      var ttlset = new TtlSet(50)
+
+      var spy = sinon.spy()
+      ttlset.on('insert', spy)
+
+      ttlset.bump(1)
+      ttlset.bump(2)
+      ttlset.bump(3)
+
+      expect(spy).to.have.been.calledThrice
+      expect(spy).to.have.been.calledWith(1)
+      expect(spy).to.have.been.calledWith(2)
+      expect(spy).to.have.been.calledWith(3)
+
+      ttlset.stop()
+      done()
+    })
+
+    it('should not emit "insert" for new entries if SILENT', function(done) {
+      var ttlset = new TtlSet(50)
+
+      var spy = sinon.spy()
+      ttlset.on('insert', spy)
+
+      ttlset.bump(1, Date.now(), TtlSet.SILENT)
+      ttlset.bump(2, Date.now())
+      ttlset.bump(3, Date.now(), TtlSet.SILENT)
+
+      expect(spy).to.have.been.calledOnce
+      expect(spy).to.have.been.calledWith(2)
+
+      ttlset.stop()
+      done()
+    })
+
     it('should create an item for the value if none exists', function(done) {
       var ttlset = new TtlSet(5000)
       ttlset.bump(5)
@@ -95,6 +131,45 @@ describe('TtlSet', function() {
 
   describe('drop', function() {
 
+    it('should emit "drop" for the dropped entry', function(done) {
+      var ttlset = new TtlSet(50)
+
+      var spy = sinon.spy()
+      ttlset.on('drop', spy)
+
+      ttlset.bump(1)
+      ttlset.bump(2)
+      ttlset.bump(3)
+      ttlset.drop(1)
+      ttlset.drop(3)
+
+      expect(spy).to.have.been.calledTwice
+      expect(spy).to.have.been.calledWith(1)
+      expect(spy).to.have.been.calledWith(3)
+
+      ttlset.stop()
+      done()
+    })
+
+    it('should not emit "drop" for the dropped entry if SILENT', function(done) {
+      var ttlset = new TtlSet(50)
+
+      var spy = sinon.spy()
+      ttlset.on('drop', spy)
+
+      ttlset.bump(1)
+      ttlset.bump(2)
+      ttlset.bump(3)
+      ttlset.drop(1, TtlSet.SILENT)
+      ttlset.drop(3)
+
+      expect(spy).to.have.been.calledOnce
+      expect(spy).to.have.been.calledWith(3)
+
+      ttlset.stop()
+      done()
+    })
+
     it('should silently ignore unknown values', function(done) {
       var ttlset = new TtlSet(5000)
       ttlset.drop(5)