diff --git a/docs/api/presence.md b/docs/api/presence.md index d8768a7a..f1ee656d 100644 --- a/docs/api/presence.md +++ b/docs/api/presence.md @@ -54,7 +54,7 @@ Optional ### unsubscribe() -Unsubscribe from presence updates from remote clients. +Unsubscribe from presence updates from remote clients. If the channel was subscribed, or a subscription was pending, any presence this client has submitted on it is broadcast to remote subscribers as `null`. ```javascript presence.unsubscribe([callback]) diff --git a/docs/middleware/actions.md b/docs/middleware/actions.md index 3d34dc6a..f44c3f42 100644 --- a/docs/middleware/actions.md +++ b/docs/middleware/actions.md @@ -78,6 +78,10 @@ This action has these additional `context` properties: `presence` -- Object > The presence object that was received. Its shape depends on its [type]({{ site.baseurl }}{% link types/index.md %}) +> +> `presence.p` is `null` when the presence leaves the channel: when it is destroyed, or when its client unsubscribes or disconnects. When its client unsubscribes or disconnects, the leave reuses the client's latest `presence` object for that ID, which passes through this middleware again with `p` set to `null` and any changes from its first pass kept, even if it is still being processed. Middleware with side effects should expect to see the same presence object twice. +> +> A presence that passes this middleware may still not be broadcast. If a newer presence with the same ID from the same client, or its leave, replaces it before it is published, its client gets an acknowledgement but nobody else receives it. ## `'sendPresence'` diff --git a/lib/agent.js b/lib/agent.js index 23092885..83a84263 100644 --- a/lib/agent.js +++ b/lib/agent.js @@ -45,14 +45,26 @@ function Agent(backend, stream) { this.presenceSubscriptionSeq = Object.create(null); // Keep track of the last request that has been sent by each local presence // belonging to this agent. This is used to generate a new disconnection - // request if the client disconnects ungracefully. This is a - // map of channel -> id -> request + // request if the client disconnects ungracefully, and a presence that's + // been replaced here by the time it would be published isn't sent. This + // is a map of channel -> id -> request this.presenceRequests = Object.create(null); // Keep track of the latest known Doc version, so that we can avoid fetching // ops to transform presence if not needed this.latestDocVersionStreams = Object.create(null); this.latestDocVersions = Object.create(null); + // Every doc, doc version and presence stream this agent holds, so that + // _cleanup() destroys them even if they've gone missing from their own map. + // Map of key -> stream + this._streams = Object.create(null); + this._nextStreamKey = 1; + // Presence broadcasts still in their doc version lookup or receivePresence + // middleware. _retractPresences() flags them to go out as a leave, and a + // leave from the client that gets through middleware flags its own older + // ones to be dropped + this._pendingPresenceBroadcasts = []; + // We need to track this manually to make sure we don't reply to messages // after the stream was closed. this.closed = false; @@ -93,20 +105,16 @@ Agent.prototype._cleanup = function() { this.backend.agentsCount--; if (!this.stream.isServer) this.backend.remoteAgentsCount--; - // Clean up doc subscription streams - for (var collection in this.subscribedDocs) { - var docs = this.subscribedDocs[collection]; - for (var id in docs) { - var stream = docs[id]; - stream.destroy(); - } + for (var channel in this.presenceRequests) { + this._retractPresences(channel); } - this.subscribedDocs = Object.create(null); - for (var channel in this.subscribedPresences) { - this.subscribedPresences[channel].destroy(); + for (var key in this._streams) { + this._streams[key].destroy(); } + this.subscribedDocs = Object.create(null); this.subscribedPresences = Object.create(null); + this.latestDocVersionStreams = Object.create(null); // Clean up query subscription streams for (var id in this.subscribedQueries) { @@ -114,12 +122,16 @@ Agent.prototype._cleanup = function() { emitter.destroy(); } this.subscribedQueries = Object.create(null); +}; - for (var collection in this.latestDocVersionStreams) { - var streams = this.latestDocVersionStreams[collection]; - for (var id in streams) streams[id].destroy(); - } - this.latestDocVersionStreams = Object.create(null); +Agent.prototype._trackStream = function(stream) { + if (stream.open === false) return; + var streams = this._streams; + var key = this._nextStreamKey++; + streams[key] = stream; + stream.once('close', function() { + delete streams[key]; + }); }; /** @@ -142,6 +154,7 @@ Agent.prototype._subscribeToStream = function(collection, id, stream) { Agent.prototype._subscribeMapToStream = function(map, collection, id, stream, dataHandler) { if (this.closed) return stream.destroy(); + this._trackStream(stream); var streams = map[collection] || (map[collection] = Object.create(null)); @@ -163,8 +176,13 @@ Agent.prototype._subscribeMapToStream = function(map, collection, id, stream, da Agent.prototype._subscribeToPresenceStream = function(channel, stream) { if (this.closed) return stream.destroy(); + this._trackStream(stream); var agent = this; + var previous = this.subscribedPresences[channel]; + if (previous) previous.destroy(); + this.subscribedPresences[channel] = stream; + stream.on('data', function(data) { if (data.error) { logger.error('Presence subscription stream error', channel, data.error); @@ -173,19 +191,34 @@ Agent.prototype._subscribeToPresenceStream = function(channel, stream) { }); stream.on('end', function() { - var requests = agent.presenceRequests[channel] || {}; - for (var id in requests) { - var request = agent.presenceRequests[channel][id]; - request.seq++; - request.p = null; - agent._broadcastPresence(request, function(error) { - if (error) logger.error('Error broadcasting disconnect presence', channel, error); - }); - } if (agent.subscribedPresences[channel] === stream) { delete agent.subscribedPresences[channel]; } - delete agent.presenceRequests[channel]; + }); +}; + +Agent.prototype._retractPresences = function(channel) { + var pendingBroadcasts = this._pendingPresenceBroadcasts; + for (var i = 0; i < pendingBroadcasts.length; i++) { + if (pendingBroadcasts[i].channel === channel) pendingBroadcasts[i].retracted = true; + } + + var requests = this.presenceRequests[channel]; + for (var id in requests) { + var request = requests[id]; + var alreadyLeft = request.p === null; + if (alreadyLeft) this._unsubscribeDocVersion(request.c, request.d, function() {}); + else this._broadcastLeave(request); + } + delete this.presenceRequests[channel]; +}; + +Agent.prototype._broadcastLeave = function(presence, callback) { + var agent = this; + presence.p = null; + this._broadcastPresence(presence, [], function(error, result) { + if (error) logger.error('Error broadcasting disconnect presence', presence.ch, presence.id, agent._src(), error); + if (callback) callback(error, result); }); }; @@ -522,7 +555,8 @@ Agent.prototype._handleMessage = function(request, callback) { message: 'Type does not support presence: ' + presence.t }); } - return this._broadcastPresence(presence, callback); + var olderBroadcasts = presence.p === null ? this._olderPendingPresenceBroadcasts(presence) : []; + return this._broadcastPresence(presence, olderBroadcasts, callback); case ACTIONS.presenceSubscribe: if (!this.backend.presenceEnabled) return; return this._subscribePresence(request.ch, request.seq, callback); @@ -843,28 +877,43 @@ Agent.prototype._src = function() { return this.src || this.clientId; }; -Agent.prototype._broadcastPresence = function(presence, callback) { +Agent.prototype._broadcastPresence = function(presence, supersededBroadcasts, callback) { var agent = this; var backend = this.backend; - var presenceRequests = this.presenceRequests; + var leaving = presence.p === null; + var pendingBroadcast = this._addPendingPresenceBroadcast(presence); var context = { presence: presence, collection: presence.c }; var start = Date.now(); - var subscriptionUpdater = presence.p === null ? + var subscriptionUpdater = leaving ? this._unsubscribeDocVersion.bind(this) : this._subscribeDocVersion.bind(this); subscriptionUpdater(presence.c, presence.d, function(error) { - if (error) return callback(error); + if (error) { + agent._removePendingPresenceBroadcast(pendingBroadcast); + return callback(error); + } backend.trigger(backend.MIDDLEWARE_ACTIONS.receivePresence, agent, context, function(error) { + agent._removePendingPresenceBroadcast(pendingBroadcast); if (error) return callback(error); - var requests = presenceRequests[presence.ch] || (presenceRequests[presence.ch] = Object.create(null)); + for (var i = 0; i < supersededBroadcasts.length; i++) { + supersededBroadcasts[i].superseded = true; + } + if (pendingBroadcast.superseded && !leaving) { + return agent._unsubscribeDocVersion(presence.c, presence.d, function() { + callback(null, presence); + }); + } + var clientHasLeft = agent.closed || pendingBroadcast.retracted; + if (clientHasLeft && !leaving) return agent._broadcastLeave(presence, callback); + var requests = agent.presenceRequests[presence.ch] || (agent.presenceRequests[presence.ch] = Object.create(null)); var previousRequest = requests[presence.id]; - if (!previousRequest || previousRequest.pv < presence.pv) { - presenceRequests[presence.ch][presence.id] = presence; + if (!previousRequest || previousRequest.p === null || previousRequest.pv < presence.pv) { + requests[presence.id] = presence; } var transformer = function(agent, presence, callback) { @@ -877,19 +926,52 @@ Agent.prototype._broadcastPresence = function(presence, callback) { transformer = backend.transformPresenceToLatestVersion.bind(backend); } - transformer(agent, presence, function(error, presence) { - if (error) return callback(error); - var channel = agent._getPresenceChannel(presence.ch); - agent.backend.pubsub.publish([channel], presence, function(error) { + transformer(agent, presence, function(error, transformedPresence) { + if (error) { + agent._withdrawPresenceRequest(requests, presence, previousRequest); + return callback(error); + } + var replaced = util.dig(agent.presenceRequests, presence.ch, presence.id) !== presence; + if (replaced && !leaving) return callback(null, transformedPresence); + var channel = agent._getPresenceChannel(transformedPresence.ch); + agent.backend.pubsub.publish([channel], transformedPresence, function(error) { if (error) return callback(error); backend.emit('timing', 'presence.broadcast', Date.now() - start, context); - callback(null, presence); + callback(null, transformedPresence); }); }); }); }); }; +Agent.prototype._withdrawPresenceRequest = function(requests, presence, previousRequest) { + if (requests[presence.id] !== presence) return; + if (previousRequest) requests[presence.id] = previousRequest; + else presence.p = null; +}; + +Agent.prototype._addPendingPresenceBroadcast = function(presence) { + var broadcast = {channel: presence.ch, id: presence.id, pv: presence.pv, retracted: false, superseded: false}; + this._pendingPresenceBroadcasts.push(broadcast); + return broadcast; +}; + +Agent.prototype._removePendingPresenceBroadcast = function(broadcast) { + var index = this._pendingPresenceBroadcasts.indexOf(broadcast); + if (index !== -1) this._pendingPresenceBroadcasts.splice(index, 1); +}; + +Agent.prototype._olderPendingPresenceBroadcasts = function(presence) { + var olderBroadcasts = []; + var pendingBroadcasts = this._pendingPresenceBroadcasts; + for (var i = 0; i < pendingBroadcasts.length; i++) { + var broadcast = pendingBroadcasts[i]; + var isOlder = broadcast.channel === presence.ch && broadcast.id === presence.id && broadcast.pv < presence.pv; + if (isOlder) olderBroadcasts.push(broadcast); + } + return olderBroadcasts; +}; + Agent.prototype._subscribeDocVersion = function(collection, id, callback) { if (!collection || !id) return callback(); @@ -957,11 +1039,6 @@ Agent.prototype._subscribePresence = function(channel, seq, cb) { return callback(); } agent.presenceSubscriptionSeq[channel] = seq; - if (agent.subscribedPresences[channel]) { - stream.destroy(); - return callback(); - } - agent.subscribedPresences[channel] = stream; agent._subscribeToPresenceStream(channel, stream); agent._requestPresence(channel, function(error) { callback(error); @@ -974,6 +1051,7 @@ Agent.prototype._unsubscribePresence = function(channel, seq, callback) { var stream = this.subscribedPresences[channel]; if (stream) stream.destroy(); delete this.subscribedPresences[channel]; + this._retractPresences(channel); callback(null, {ch: channel, seq: seq}); }; diff --git a/test/agent.js b/test/agent.js index b364b7df..e2fb1fea 100644 --- a/test/agent.js +++ b/test/agent.js @@ -8,6 +8,10 @@ var ACTIONS = require('../lib/message-actions').ACTIONS; var Connection = require('../lib/client/connection'); var protocol = require('../lib/protocol'); var LegacyConnection = require('sharedb-legacy/lib/client').Connection; +var async = require('async'); +var types = require('../lib/types'); +var presenceTestType = require('./client/presence/presence-test-type'); +types.register(presenceTestType.type); describe('Agent', function() { var backend; @@ -323,4 +327,124 @@ describe('Agent', function() { }); }); }); + + describe('cleanup', function() { + var connection; + + beforeEach(function(done) { + backend.close(function(error) { + if (error) return done(error); + backend = new Backend({presence: true}); + connection = backend.connect(); + connection.get('books', 'northern-lights').create('North Lights', presenceTestType.type.name, done); + }); + }); + + function closeConnection(callback) { + connection.agent.stream.once('end', callback); + connection.close(); + } + + function opensStreams(count, step) { + return function(next) { + var streamsCount = backend.pubsub.streamsCount; + step(function(error) { + if (error) return next(error); + expect(backend.pubsub.streamsCount - streamsCount).to.equal(count); + next(); + }); + }; + } + + it('destroys a stream that is missing from its subscription map', function(done) { + connection.get('dogs', 'fido').subscribe(function(error) { + if (error) return done(error); + delete connection.agent.subscribedDocs.dogs; + closeConnection(function() { + expect(backend.pubsub.streamsCount).to.equal(0); + done(); + }); + }); + }); + + it('stops tracking streams once they close', function(done) { + var doc = connection.get('dogs', 'fido'); + var presence = connection.getDocPresence('books', 'northern-lights'); + var localPresence = presence.create('presence-1'); + async.series([ + doc.subscribe.bind(doc), + doc.unsubscribe.bind(doc), + presence.subscribe.bind(presence), + presence.unsubscribe.bind(presence), + localPresence.submit.bind(localPresence, {index: 0}), + localPresence.destroy.bind(localPresence), + function(next) { + expect(connection.agent._streams).to.be.empty; + next(); + } + ], done); + }); + + it('empties its stream maps as soon as the connection closes', function(done) { + var presence = connection.getDocPresence('books', 'northern-lights'); + var agent = connection.agent; + async.series([ + function(next) { + connection.get('dogs', 'fido').subscribe(next); + }, + presence.subscribe.bind(presence), + function(next) { + presence.create('presence-1').submit({index: 0}, next); + }, + function(next) { + agent.stream.once('end', function() { + expect(agent.subscribedDocs).to.be.empty; + expect(agent.subscribedPresences).to.be.empty; + expect(agent.latestDocVersionStreams).to.be.empty; + next(); + }); + connection.close(); + } + ], done); + }); + + it('does not track a stream that closed before reaching the agent', function(done) { + backend.use(backend.MIDDLEWARE_ACTIONS.readSnapshots, function(context, next) { + backend.pubsub.close(); + next(); + }); + connection.get('dogs', 'fido').subscribe(function(error) { + if (error) return done(error); + expect(connection.agent._streams).to.be.empty; + done(); + }); + }); + + it('destroys every pub/sub stream when the connection closes', function(done) { + var presence = connection.getDocPresence('books', 'northern-lights'); + async.series([ + opensStreams(1, function(next) { + connection.get('dogs', 'fido').subscribe(next); + }), + opensStreams(2, function(next) { + connection.startBulk(); + connection.get('dogs', 'rex').subscribe(); + connection.get('dogs', 'spot').subscribe(next); + connection.endBulk(); + }), + opensStreams(1, presence.subscribe.bind(presence)), + opensStreams(1, function(next) { + presence.create('presence-1').submit({index: 0}, next); + }), + opensStreams(1, function(next) { + connection.createSubscribeQuery('dogs', {}, null, next); + }), + closeConnection, + function(next) { + expect(backend.pubsub.streamsCount).to.equal(0); + next(); + } + ], done); + }); + }); }); diff --git a/test/client/presence/doc-presence.js b/test/client/presence/doc-presence.js index 5cb4b5ef..7b9b3828 100644 --- a/test/client/presence/doc-presence.js +++ b/test/client/presence/doc-presence.js @@ -793,6 +793,21 @@ describe('DocPresence', function() { ], done); }); + it('does not keep presence when fetching its doc version fails', function(done) { + var localPresence1 = presence1.create('presence-1'); + + backend.use(backend.MIDDLEWARE_ACTIONS.readSnapshots, function(context, next) { + next(new Error('denied')); + }); + + localPresence1.submit({index: 1}, function(error) { + expect(error.message).to.contain('denied'); + expect(connection1.agent.presenceRequests).to.be.empty; + expect(connection1.agent._pendingPresenceBroadcasts).to.be.empty; + done(); + }); + }); + it('errors when trying to submit presence on a type that does not support it', function(done) { var jsonDoc = connection1.get('books', 'snuff'); var jsonPresence = connection1.getDocPresence('books', 'snuff'); @@ -1047,6 +1062,252 @@ describe('DocPresence', function() { ], done); }); + it('releases the doc version stream of presence in flight when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + var streamsCount; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + presence1.unsubscribe(function(error) { + if (error) return done(error); + next(); + }); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + streamsCount = backend.pubsub.streamsCount; + next(); + }, + presence1.subscribe.bind(presence1), + function(next) { + async.parallel([ + function(cb) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + cb(); + }); + }, + function(cb) { + localPresence1.submit({index: 1}, cb); + } + ], next); + }, + function(next) { + expect(backend.pubsub.streamsCount).to.equal(streamsCount); + next(); + } + ], done); + }); + + it('releases the doc version stream of presence transformed across a delete when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence1.subscribe.bind(presence1), + doc1.unsubscribe.bind(doc1), + doc2.del.bind(doc2), + localPresence1.submit.bind(localPresence1, {index: 1}), + function(next) { + expect(connection1.agent.presenceRequests['books.northern-lights']['presence-1'].p).to.be.null; + expect(connection1.agent.latestDocVersionStreams).not.to.be.empty; + presence1.unsubscribe(next); + }, + function(next) { + expect(connection1.agent.latestDocVersions).to.be.empty; + expect(connection1.agent.latestDocVersionStreams).to.be.empty; + next(); + } + ], done); + }); + + it('releases the doc version stream of presence that failed to transform when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + doc1.unsubscribe.bind(doc1), + doc2.submitOp.bind(doc2, {index: 5, value: 'ern'}), + function(next) { + localPresence1.submit({badProp: 'foo'}, function(error) { + expect(error.code).to.equal('ERR_PRESENCE_TRANSFORM_FAILED'); + expect(connection1.agent.latestDocVersionStreams).not.to.be.empty; + presence1.unsubscribe(next); + }); + }, + function(next) { + expect(connection1.agent.latestDocVersions).to.be.empty; + expect(connection1.agent.latestDocVersionStreams).to.be.empty; + localPresence2.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-2'); + next(); + }); + } + ], done); + }); + + it('does not broadcast an older presence that finishes transforming after its client unsubscribes', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + var releaseFirstTransform; + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + doc1.unsubscribe.bind(doc1), + doc2.submitOp.bind(doc2, {index: 5, value: 'ern'}), + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.op, function(context, callback) { + if (releaseFirstTransform || context.agent !== connection1.agent) return callback(); + releaseFirstTransform = callback; + next(); + }); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + localPresence1.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + presence1.unsubscribe(errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + releaseFirstTransform(); + localPresence2.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-2'); + next(); + }); + }, + function(next) { + expect(presence2.remotePresences).to.have.all.keys('presence-2'); + next(); + } + ], done); + }); + + it('broadcasts an older presence still transforming when a newer one fails to transform', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releaseFirstTransform; + + async.series([ + presence2.subscribe.bind(presence2), + doc1.unsubscribe.bind(doc1), + doc2.submitOp.bind(doc2, {index: 5, value: 'ern'}), + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.op, function(context, callback) { + if (releaseFirstTransform || context.agent !== connection1.agent) return callback(); + releaseFirstTransform = callback; + next(); + }); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + localPresence1.submit({badProp: 'foo'}, function(error) { + expect(error.code).to.equal('ERR_PRESENCE_TRANSFORM_FAILED'); + next(); + }); + }, + function(next) { + releaseFirstTransform(); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 1}); + next(); + }); + } + ], done); + }); + + it('broadcasts a null presence still being transformed when the connection closes', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + doc1.unsubscribe.bind(doc1), + doc2.submitOp.bind(doc2, {index: 5, value: 'ern'}), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.op, function(context, callback) { + if (context.agent !== connection1.agent) return callback(); + connection1.agent.stream.once('end', function() { + callback(); + }); + connection1.close(); + }); + localPresence1.submit(null, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + } + ], done); + }); + + it('does not broadcast presence still looking up its doc version when it is destroyed', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + var releaseDocVersion; + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.readSnapshots, function(context, callback) { + if (releaseDocVersion || context.agent !== connection1.agent) return callback(); + releaseDocVersion = callback; + next(); + }); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + localPresence1.destroy(errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + connection1.once('receive', function(request) { + expect(request.data).to.include({id: 'presence-1', pv: 0}); + expect(connection1.agent.latestDocVersions).to.be.empty; + expect(connection1.agent.latestDocVersionStreams).to.be.empty; + next(); + }); + releaseDocVersion(); + }, + function(next) { + localPresence2.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-2'); + next(); + }); + }, + function(next) { + expect(presence2.remotePresences).to.have.all.keys('presence-2'); + next(); + } + ], done); + }); + it('does not error when destroying presence for a deleted doc', function(done) { var localPresence1 = presence1.create('presence-1'); diff --git a/test/client/presence/presence.js b/test/client/presence/presence.js index f972327e..cd7d321e 100644 --- a/test/client/presence/presence.js +++ b/test/client/presence/presence.js @@ -1,4 +1,5 @@ var Backend = require('../../../lib/backend'); +var logger = require('../../../lib/logger'); var PresencePauser = require('./presence-pauser'); var expect = require('chai').expect; var async = require('async'); @@ -718,6 +719,513 @@ describe('Presence', function() { ], done); }); + it('broadcasts a null presence when a connection that never subscribed is disconnected', function(done) { + var localPresence1 = presence1.create('presence-1'); + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + connection1.close(); + } + ], done); + }); + + it('broadcasts a null presence for presence in flight when the connection is disconnected', function(done) { + var localPresence1 = presence1.create('presence-1'); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + connection1.agent.stream.once('end', function() { + next(); + }); + connection1.close(); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + localPresence1.submit({index: 3}, errorHandler(done)); + } + ], done); + }); + + it('broadcasts a null presence for presence in flight when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + presence1.unsubscribe(function(error) { + if (error) return done(error); + next(); + }); + }); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + async.parallel([ + function(cb) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + cb(); + }); + }, + function(cb) { + localPresence1.submit({index: 3}, cb); + } + ], next); + } + ], done); + }); + + it('broadcasts presence submitted after unsubscribing while earlier presence was in flight', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + var received = {}; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.id !== 'presence-1' || context.presence.p === null) return next(); + presence1.unsubscribe(function(error) { + if (error) return done(error); + localPresence2.submit({index: 2}, function(error) { + if (error) return done(error); + next(); + }); + }); + }); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + presence2.on('receive', function(id, presence) { + received[id] = presence; + if (Object.keys(received).length === 2) next(); + }); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + expect(received).to.eql({'presence-1': null, 'presence-2': {index: 2}}); + next(); + } + ], done); + }); + + it('broadcasts a null presence for presence held in receive middleware when disconnected', function(done) { + var localPresence1 = presence1.create('presence-1'); + + backend.use(backend.MIDDLEWARE_ACTIONS.receive, function(context, next) { + if (context.agent !== connection1.agent || context.data.a !== 'p') return next(); + connection1.agent.stream.once('end', function() { + next(); + }); + connection1.close(); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + localPresence1.submit({index: 3}, errorHandler(done)); + } + ], done); + }); + + it('does not rewrite a null presence again when middleware fills it in on disconnect', function(done) { + var localPresence1 = presence1.create('presence-1'); + var nullPresenceCount = 0; + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, callback) { + if (context.presence.p !== null) return callback(); + if (++nullPresenceCount > 1) return done(new Error('Rewrote a null presence more than once')); + context.presence.p = {index: 0}; + callback(); + }); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-1'); + expect(nullPresenceCount).to.equal(1); + next(); + }); + connection1.close(); + } + ], done); + }); + + it('logs an error when the leave for presence in flight fails', function(done) { + var localPresence1 = presence1.create('presence-1'); + var agent1 = connection1.agent; + + sinon.stub(logger, 'error').callsFake(function(message, channel, id, src, error) { + expect(message).to.equal('Error broadcasting disconnect presence'); + expect([channel, id, src]).to.eql(['test-channel', 'presence-1', agent1._src()]); + expect(error.message).to.equal('bad!'); + done(); + }); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(new Error('bad!')); + agent1.stream.once('end', function() { + next(); + }); + connection1.close(); + }); + + localPresence1.submit({index: 3}, errorHandler(done)); + }); + + it('does not broadcast presence that gets through middleware after its own leave', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + var releasePresence; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + var isFirstPresence = context.presence.id === 'presence-1' && context.presence.pv === 0; + if (!isFirstPresence || releasePresence) return next(); + releasePresence = next; + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + localPresence1.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + localPresence1.submit(null, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + releasePresence(); + localPresence2.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function(id) { + expect(id).to.equal('presence-2'); + next(); + }); + }, + function(next) { + expect(presence2.remotePresences).to.eql({'presence-2': {index: 3}}); + next(); + } + ], done); + }); + + it('broadcasts presence in flight when its leave is rejected', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releasePresence; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(new Error('bad!')); + releasePresence = next; + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + localPresence1.submit(null, function(error) { + expect(error.message).to.contain('bad!'); + next(); + }); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 1}); + next(); + }); + releasePresence(); + } + ], done); + }); + + it('broadcasts presence that overtakes its older leave in receive middleware', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releaseLeave; + var releasePresence; + + backend.use(backend.MIDDLEWARE_ACTIONS.receive, function(context, next) { + var isLeave = context.agent === connection1.agent && context.data.a === 'p' && context.data.p === null; + if (!isLeave) return next(); + releaseLeave = next; + }); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + releasePresence = next; + releaseLeave(); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + localPresence1.submit(null, errorHandler(done)); + localPresence1.submit({index: 1}, errorHandler(done)); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 1}); + next(); + }); + releasePresence(); + } + ], done); + }); + + it('broadcasts presence re-created with the same id while its old leave is in middleware', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releaseLeave; + var releasePresence; + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, callback) { + if (releaseLeave && releasePresence) return callback(); + if (context.presence.p === null) releaseLeave = callback; + else releasePresence = callback; + if (releaseLeave && releasePresence) next(); + }); + localPresence1.submit(null); + presence1.create('presence-1').submit({index: 2}, errorHandler(done)); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + releaseLeave(); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 2}); + next(); + }); + releasePresence(); + } + ], done); + }); + + it('broadcasts presence submitted after unsubscribing while an older one is still in middleware', function(done) { + var localPresence1 = presence1.create('presence-1'); + var heldPresences = []; + var onHeld; + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.p === null) return next(); + heldPresences.push(next); + onHeld(); + }); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + onHeld = next; + localPresence1.submit({index: 1}, errorHandler(done)); + }, + presence1.unsubscribe.bind(presence1), + function(next) { + onHeld = next; + localPresence1.submit({index: 2}, errorHandler(done)); + }, + function(next) { + heldPresences[0](); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + heldPresences[1](); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 2}); + next(); + }); + } + ], done); + }); + + it('broadcasts presence re-created with the same id, and its leave on disconnect', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + localPresence1.destroy(errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + next(); + }); + }, + function(next) { + presence1.create('presence-1').submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + connection1.close(); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + } + ], done); + }); + + it('broadcasts presence re-created with the same id after destroying the channel presence', function(done) { + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + presence1.create('presence-1').submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + var leaveCount = 0; + presence1.destroy(errorHandler(done)); + presence2.on('receive', function receiveLeave(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + leaveCount++; + if (leaveCount < 2) return; + presence2.off('receive', receiveLeave); + next(); + }); + }, + function(next) { + connection1.getPresence('test-channel').create('presence-1').submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + connection1.close(); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + } + ], done); + }); + + it('broadcasts a null presence when unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + presence1.unsubscribe(errorHandler(done)); + } + ], done); + }); + + it('does not broadcast a null presence again when disconnecting after unsubscribing', function(done) { + var localPresence1 = presence1.create('presence-1'); + var agent1 = connection1.agent; + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 3}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + presence2.once('receive', function() { + next(); + }); + presence1.unsubscribe(errorHandler(done)); + }, + function(next) { + sinon.spy(agent1, '_broadcastPresence'); + agent1.stream.once('end', next); + connection1.close(); + }, + function(next) { + expect(agent1._broadcastPresence).not.to.have.been.called; + next(); + } + ], done); + }); + describe('middleware', function() { describe('receivePresence', function() { it('provides the presence in the middleware', function(done) { @@ -732,7 +1240,7 @@ describe('Presence', function() { it('can mutate the presence in the middleware', function(done) { backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { - context.presence.p.index++; + if (context.presence.p) context.presence.p.index++; next(); }); @@ -767,8 +1275,34 @@ describe('Presence', function() { localPresence1.submit({index: 5}, function(error) { expect(error.message).to.contain('bad!'); expect(connection1.agent.presenceRequests['test-channel']).not.to.be.ok; + expect(connection1.agent._pendingPresenceBroadcasts).to.be.empty; + next(); + }); + } + ], done); + }); + + it('still broadcasts presence in flight when another presence on the channel is rejected', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + + backend.use(backend.MIDDLEWARE_ACTIONS.receivePresence, function(context, next) { + if (context.presence.id === 'presence-2') return next(new Error('bad!')); + localPresence2.submit({index: 2}, function(error) { + expect(error.message).to.contain('bad!'); + next(); + }); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 1}); next(); }); + localPresence1.submit({index: 1}, errorHandler(done)); } ], done); });