diff --git a/lib/db/api.js b/lib/db/api.js index 90e30f4a..ad5c6c91 100644 --- a/lib/db/api.js +++ b/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() })) } diff --git a/lib/units/processor/index.js b/lib/units/processor/index.js index 216fd44d..edb4555e 100644 --- a/lib/units/processor/index.js +++ b/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]) diff --git a/lib/units/provider/index.js b/lib/units/provider/index.js index aedef773..3e4fd919 100644 --- a/lib/units/provider/index.js +++ b/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( diff --git a/lib/units/reaper/index.js b/lib/units/reaper/index.js index c99eb690..0c961b23 100644 --- a/lib/units/reaper/index.js +++ b/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,7 +83,7 @@ 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) }) }) }) @@ -81,14 +91,14 @@ module.exports = function(options) { 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()) } diff --git a/lib/units/websocket/index.js b/lib/units/websocket/index.js index 39f67bc1..194a9980 100644 --- a/lib/units/websocket/index.js +++ b/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' } }) diff --git a/lib/util/ttlset.js b/lib/util/ttlset.js index 08ebcdcd..408dd7d0 100644 --- a/lib/util/ttlset.js +++ b/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) + } } } diff --git a/lib/wire/wire.proto b/lib/wire/wire.proto index 2d0e3825..8113e4e1 100644 --- a/lib/wire/wire.proto +++ b/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; } diff --git a/test/util/ttlset.js b/test/util/ttlset.js index 6c0f418e..a6b510d5 100644 --- a/test/util/ttlset.js +++ b/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)