Reduce messages when lots of participants join at same time (#877)

* Reduce messages when lots of participants join at same time

Author: starwarfan

source/agent/conference/conference.js

@@ -528,16 +528,18 @@ var Conference = function (rpcClient, selfRpcId) {
 
   const sendMsg = function(from, to, msg, data) {
     log.debug('sendMsg, from:', from, 'to:', to, 'msg:', msg, 'data:', data);
-    if (to === 'all') {
-      for (var participant_id in participants) {
-        sendMsgTo(participant_id, msg, data);
+    if (to === 'all' || to === 'others') {
+      // Broadcast message to portal
+      let excludes = (to === 'others') ? [from] : [];
+      let portals = new Set();
+      for (let pptId in participants) {
+        portals.add(participants[pptId].getPortal());
       }
-    } else if (to === 'others') {
-      for (var participant_id in participants) {
-        if (participant_id !== from) {
-          sendMsgTo(participant_id, msg, data);
+      portals.forEach((portal) => {
+        if (portal) {
+          rpcReq.broadcast(portal, selfRpcId, excludes, msg, data);
         }
-      }
+      });
     } else {
       sendMsgTo(to, msg, data);
     }

source/agent/conference/rpcRequest.js

@@ -66,6 +66,10 @@ var RpcRequest = function(rpcChannel) {
     return rpcChannel.makeRPC(portal, 'notify', [participantId, event, data]);
   };
 
+  that.broadcast = function(portal, controller, excludeList, event, data) {
+    return rpcChannel.makeRPC(portal, 'broadcast', [controller, excludeList, event, data]);
+  };
+
   that.dropUser = function(portal, participantId) {
     return rpcChannel.makeRPC(portal, 'drop', [participantId]);
   };

source/agent/conference/test/rpcRequest_test.js

@@ -134,7 +134,7 @@ describe('rpcRequest.terminate', function() {
    });
 });
 
-describe('rpcRequest.getRoomConfig/onSessionSignaling/mediaOnOff/sendMsg/dropUser', function() {
+describe('rpcRequest.getRoomConfig/onSessionSignaling/mediaOnOff/sendMsg/broadcast/dropUser', function() {
   it('Should succeed if rpcChannel.makeRPC succeeds.', function() {
     var mockRpcChannel = sinon.createStubInstance(rpcChannel);
     mockRpcChannel.makeRPC = sinon.stub();
@@ -145,6 +145,7 @@ describe('rpcRequest.getRoomConfig/onSessionSignaling/mediaOnOff/sendMsg/dropUse
     var mediaOnOff = req.mediaOnOff('rpcIdOfWorkerNode', 'sessionId', 'video', 'in', 'off');
     var onSessionSignaling = req.onSessionSignaling('rpcIdOfWorkerNode', 'sessionId', 'soacObj');
     var sendMsg = req.sendMsg('rpcIdOfPortal', 'participantId', 'event-name', 'msgObj');
+    var broadcast = req.broadcast('rpcIdOfPortal', 'controller', ['participantId'], 'event-name', 'msgObj');
     var dropUser = req.dropUser('rpcIdOfPortal', 'participantId');
 
 
@@ -153,10 +154,11 @@ describe('rpcRequest.getRoomConfig/onSessionSignaling/mediaOnOff/sendMsg/dropUse
       expect(mediaOnOff).to.become('ok-or-data'),
       expect(onSessionSignaling).to.become('ok-or-data'),
       expect(sendMsg).to.become('ok-or-data'),
+      expect(broadcast).to.become('ok-or-data'),
       expect(dropUser).to.become('ok-or-data')
       ])
       .then(function() {
-        expect(mockRpcChannel.makeRPC.callCount).to.equal(5);
+        expect(mockRpcChannel.makeRPC.callCount).to.equal(6);
       });
   });
 
@@ -170,17 +172,19 @@ describe('rpcRequest.getRoomConfig/onSessionSignaling/mediaOnOff/sendMsg/dropUse
     var mediaOnOff = req.mediaOnOff('rpcIdOfWorkerNode', 'sessionId', 'video', 'in', 'off');
     var onSessionSignaling = req.onSessionSignaling('rpcIdOfWorkerNode', 'sessionId', 'soacObj');
     var sendMsg = req.sendMsg('rpcIdOfPortal', 'participantId', 'event-name', 'msgObj');
+    var broadcast = req.broadcast('rpcIdOfPortal', 'controller', ['participantId'], 'event-name', 'msgObj');
     var dropUser = req.dropUser('rpcIdOfPortal', 'participantId');
 
     return Promise.all([
       expect(getRoomConfig).to.be.rejectedWith('timeout-or-error'),
       expect(mediaOnOff).to.be.rejectedWith('timeout-or-error'),
       expect(onSessionSignaling).to.be.rejectedWith('timeout-or-error'),
       expect(sendMsg).to.be.rejectedWith('timeout-or-error'),
+      expect(broadcast).to.be.rejectedWith('timeout-or-error'),
       expect(dropUser).to.be.rejectedWith('timeout-or-error')
       ])
       .then(function() {
-        expect(mockRpcChannel.makeRPC.callCount).to.equal(5);
+        expect(mockRpcChannel.makeRPC.callCount).to.equal(6);
       });
   });
 });

source/portal/index.js

@@ -241,6 +241,10 @@ var rpcPublic = {
     } else {
       callback('callback', 'error', 'Invalid token for WebTransport.');
     }
+  },
+  broadcast: function(controller, excludeList, event, data, callback) {
+    socketio_server && socketio_server.broadcast(controller, excludeList, event, data);
+    callback('callback', 'ok');
   }
 };
 

source/portal/socketIOServer.js

@@ -455,6 +455,18 @@ var SocketIOServer = function(spec, portal, observer) {
     }
   };
 
+  that.broadcast = function(controller, excludeList, event, data) {
+    log.debug('broadcast controller:', controller, 'exclude:', excludeList, 'event:', event, 'data:', data);
+    portal.getParticipantsByController('node', controller)
+      .then(function (receivers) {
+        for (let clientId of receivers) {
+          if (!excludeList.includes(clientId)) {
+            clients[clientId].notify(event, data);
+          }
+        }
+      });
+  }
+
   that.drop = function(participantId) {
     if (participantId === 'all') {
       for(var pid in clients) {

source/portal/test/socketIOServer_v10Client_test.js

@@ -744,6 +744,27 @@ describe('Notifying events to clients.', function() {
       });
     });
   });
+
+  it('Broadcasting the joined clients should succeed.', function(done) {
+    var client = sioClient.connect(serverUrl, {reconnect: false, secure: false, 'force new connection': true});
+
+    client.on('broadcast-event', function(data) {
+      expect(data).to.equal('broadcast-data');
+      done();
+    });
+
+    client.on('connect', function() {
+      mockPortal.join = sinon.stub();
+      mockPortal.join.resolves(JSON.parse(JSON.stringify(presenter_join_result)));
+
+      client.emit('login', jsLoginInfo, function(status, resp) {
+        mockPortal.join = null;
+        expect(status).to.equal('ok');
+        server.broadcast(['excludeClient'], 'broadcast-event', 'broadcast-data');
+      });
+    });
+  });
+
 });
 
 describe('Drop users from sessions.', function() {