Merge pull request #629 from share/subscription-stream-leak

⚡️ Fix Presence `Stream` leak
share · Oct 10, 2023 · e7e77ba · e7e77ba
2 parents dc033c2 + e73b26f
commit e7e77ba
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 4 deletions.
diff --git a/lib/agent.js b/lib/agent.js
@@ -809,6 +809,19 @@ Agent.prototype._createPresence = function(request) {
 
 Agent.prototype._subscribePresence = function(channel, seq, callback) {
   var agent = this;
+
+  function requestPresence() {
+    agent._requestPresence(channel, function(error) {
+      callback(error, {ch: channel, seq: seq});
+    });
+  }
+
+  var existingStream = agent.subscribedPresences[channel];
+  if (existingStream) {
+    agent.presenceSubscriptionSeq[channel] = seq;
+    return requestPresence();
+  }
+
   var presenceChannel = this._getPresenceChannel(channel);
   this.backend.pubsub.subscribe(presenceChannel, function(error, stream) {
     if (error) return callback(error);
@@ -819,17 +832,15 @@ Agent.prototype._subscribePresence = function(channel, seq, callback) {
     agent.presenceSubscriptionSeq[channel] = seq;
     agent.subscribedPresences[channel] = stream;
     agent._subscribeToPresenceStream(channel, stream);
-    agent._requestPresence(channel, function(error) {
-      callback(error, {ch: channel, seq: seq});
-    });
+    requestPresence();
   });
 };
 
 Agent.prototype._unsubscribePresence = function(channel, seq, callback) {
-  if (seq < this.presenceSubscriptionSeq[channel]) return;
   this.presenceSubscriptionSeq[channel] = seq;
   var stream = this.subscribedPresences[channel];
   if (stream) stream.destroy();
+  delete this.subscribedPresences[channel];
   callback(null, {ch: channel, seq: seq});
 };
 

diff --git a/test/client/presence/presence.js b/test/client/presence/presence.js
@@ -441,6 +441,22 @@ describe('Presence', function() {
     ], done);
   });
 
+  it('does not leak Streams when subscribing the same presence multiple times', function(done) {
+    var streamsCount;
+    async.series([
+      presence1.subscribe.bind(presence1, {force: true}),
+      function(next) {
+        streamsCount = backend.pubsub.streamsCount;
+        next();
+      },
+      presence1.subscribe.bind(presence1, {force: true}),
+      function(next) {
+        expect(backend.pubsub.streamsCount).to.equal(streamsCount);
+        next();
+      }
+    ], done);
+  });
+
   it('throws an error when trying to create a presence with a non-string ID', function() {
     expect(function() {
       presence1.create(123);