diff --git a/docs/api/presence.md b/docs/api/presence.md index d8768a7a9..f1ee656d4 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 48e7f427c..d00e5a3c9 100644 --- a/docs/middleware/actions.md +++ b/docs/middleware/actions.md @@ -78,6 +78,8 @@ 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. If its client leaves while a presence is still in this middleware, the presence passes through it a second time with `p: null`. ## `'sendPresence'` diff --git a/lib/agent.js b/lib/agent.js index 9060741f1..b7b736490 100644 --- a/lib/agent.js +++ b/lib/agent.js @@ -54,6 +54,10 @@ function Agent(backend, stream) { this.latestDocVersionStreams = Object.create(null); this.latestDocVersions = Object.create(null); + this._streams = Object.create(null); + this._nextStreamKey = 1; + 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; @@ -94,20 +98,13 @@ 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.subscribedPresences = Object.create(null); // Clean up query subscription streams for (var id in this.subscribedQueries) { @@ -115,12 +112,15 @@ 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) { + var streams = this._streams; + var key = this._nextStreamKey++; + streams[key] = stream; + stream.once('close', function() { + delete streams[key]; + }); }; /** @@ -143,6 +143,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)); @@ -164,8 +165,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); @@ -174,22 +180,30 @@ 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]; + if (request.p === null) continue; + request.p = null; + this._broadcastPresence(request, function(error) { + if (error) logger.error('Error broadcasting disconnect presence', channel, error); + }); + } + delete this.presenceRequests[channel]; +}; + Agent.prototype._subscribeToQuery = function(emitter, queryId, collection, query) { var previous = this.subscribedQueries[queryId]; if (previous) previous.destroy(); @@ -820,25 +834,34 @@ Agent.prototype._src = function() { Agent.prototype._broadcastPresence = function(presence, callback) { var agent = this; var backend = this.backend; - var presenceRequests = this.presenceRequests; + var leaving = presence.p === null; + var pendingBroadcast = this._addPendingPresenceBroadcast(presence.ch); 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)); + if ((agent.closed || pendingBroadcast.retracted) && !leaving) { + presence.p = null; + return agent._broadcastPresence(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; + requests[presence.id] = presence; } var transformer = function(agent, presence, callback) { @@ -864,6 +887,17 @@ Agent.prototype._broadcastPresence = function(presence, callback) { }); }; +Agent.prototype._addPendingPresenceBroadcast = function(channel) { + var broadcast = {channel: channel, retracted: 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._subscribeDocVersion = function(collection, id, callback) { if (!collection || !id) return callback(); @@ -931,11 +965,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); @@ -948,6 +977,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 5ee3f25a9..2b2e8221a 100644 --- a/test/agent.js +++ b/test/agent.js @@ -7,6 +7,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; @@ -187,4 +191,71 @@ 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('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 f29d5e830..541982df8 100644 --- a/test/client/presence/doc-presence.js +++ b/test/client/presence/doc-presence.js @@ -792,6 +792,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'); @@ -993,6 +1008,46 @@ 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('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 f972327e2..79cab8715 100644 --- a/test/client/presence/presence.js +++ b/test/client/presence/presence.js @@ -718,6 +718,220 @@ 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('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 +946,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 +981,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); });