Fix unexpected behaviour with single pc client (#696)

Author: starwarfan

source/agent/analytics/videoGstPipeline/binding.pipeline.gyp

@@ -7,6 +7,7 @@
       './VideoGstAnalyzer.cpp',
       '../../../core/owt_base/MediaFramePipeline.cpp',
       '../../../core/owt_base/RawTransport.cpp',
+      '../../../core/common/IOService.cpp',
       './GstInternalOut.cpp',
       './GstInternalIn.cpp',
     ],

source/agent/conference/conference.js

@@ -578,9 +578,14 @@ var Conference = function (rpcClient, selfRpcId) {
   const addStream = (id, locality, media, info) => {
     info.origin = streams[id] ? streams[id].info.origin : {isp:"isp", region:"region"};
     if (info.analytics && subscriptions[info.analytics]) {
-      const sourceTrack = subscriptions[info.analytics].media.tracks
-        .find(t => t.type === 'video');
-      const sourceId = streams[sourceTrack.from] ? sourceTrack.from : trackOwners[sourceTrack.from];
+      let sourceId;
+      if (subscriptions[info.analytics].isInConnecting) {
+        sourceId = subscriptions[info.analytics].media.video.from;
+      } else {
+        const sourceTrack = subscriptions[info.analytics].media.tracks
+            .find(t => t.type === 'video');
+        sourceId = streams[sourceTrack.from] ? sourceTrack.from : trackOwners[sourceTrack.from];
+      }
       if (streams[sourceId]) {
         info.origin = streams[sourceId].info.origin;
       } else {
@@ -1121,6 +1126,9 @@ var Conference = function (rpcClient, selfRpcId) {
   };
 
   const isResolutionAvailable = (streamVideo, resolution) => {
+    if (!streamVideo.parameters || !streamVideo.parameters.resolution) {
+      return true;
+    }
     if (streamVideo.parameters && streamVideo.parameters.resolution && isResolutionEqual(streamVideo.parameters.resolution, resolution)) {
       return true;
     }
@@ -1213,29 +1221,31 @@ var Conference = function (rpcClient, selfRpcId) {
       return callback('callback', 'error', 'Participant has not joined');
     }
 
-    if ((subDesc.media.audio && !participants[participantId].isSubscribePermitted('audio'))
-        || (subDesc.media.video && !participants[participantId].isSubscribePermitted('video'))) {
-      return callback('callback', 'error', 'unauthorized');
+    let audioTrack = null;
+    let videoTrack = null;
+    if (subDesc.type === 'webrtc') {
+      audioTrack = subDesc.media.tracks.find(t => t.type === 'audio');
+      videoTrack = subDesc.media.tracks.find(t => t.type === 'video');
+    } else {
+      audioTrack = subDesc.media.audio;
+      videoTrack = subDesc.media.video;
     }
-    if (subDesc.type === 'webrtc' && subDesc.media.tracks) {
-      if ((subDesc.media.tracks.find(t => t.type === 'audio')
-        && !participants[participantId].isSubscribePermitted('audio'))
-        || (subDesc.media.tracks.find(t => t.type === 'video')
-        && !participants[participantId].isSubscribePermitted('video'))) {
-        return callback('callback', 'error', 'unauthorized');
-      }
+
+    if ((audioTrack && !participants[participantId].isSubscribePermitted('audio'))
+        || (videoTrack && !participants[participantId].isSubscribePermitted('video'))) {
+      return callback('callback', 'error', 'unauthorized');
     }
 
     if (subscriptions[subscriptionId]) {
       return callback('callback', 'error', 'Subscription exists');
     }
 
     var requestError = { message: '' };
-    if (subDesc.media.audio && !validateAudioRequest(subDesc.type, subDesc.media.audio, requestError)) {
+    if (audioTrack && !validateAudioRequest(subDesc.type, audioTrack, requestError)) {
       return callback('callback', 'error', 'Target audio stream does NOT satisfy:' + requestError.message);
     }
 
-    if (subDesc.media.video && !validateVideoRequest(subDesc.type, subDesc.media.video, requestError)) {
+    if (videoTrack && !validateVideoRequest(subDesc.type, videoTrack, requestError)) {
       return callback('callback', 'error', 'Target video stream does NOT satisfy:' + requestError.message);
     }
 
@@ -1253,9 +1263,10 @@ var Conference = function (rpcClient, selfRpcId) {
         const rtcSubInfo = translateRtcSubIfNeeded(subDesc);
         // Set formatPreference
         rtcSubInfo.tracks.forEach(track => {
-          const source = streams[track.from].media.tracks.find(t => t.type === track.type);
+          const streamId = streams[track.from]? track.from : trackOwners[track.from];
+          const source = getStreamTrack(track.from, track.type);
           const formatPreference = {};
-          if (streams[track.from].type === 'forward') {
+          if (streams[streamId].type === 'forward') {
             formatPreference.preferred = source.format;
             source.optional && source.optional.format && (formatPreference.optional = source.optional.format);
           } else {

source/agent/conference/rtcController.js

@@ -313,25 +313,12 @@ class RtcController extends EventEmitter {
         const abortData = { direction: operation.direction, owner, reason };
         this.emit('session-aborted', sessionId, abortData);
         this.operations.delete(sessionId);
-
-        let emptyTransport = true;
-        for (const [id, op] of this.operations) {
-          if (op.transportId === transport.id) {
-            emptyTransport = false;
-            break;
-          }
-        }
-        if (emptyTransport) {
-          log.debug(`to recycleWorkerNode: ${locality} task:, ${sessionId}`);
-          const taskConfig = {room: this.roomId, task: sessionId};
-          return this.rpcReq.recycleWorkerNode(locality.agent, locality.node, taskConfig)
-            .catch(reason => {
-              log.debug(`AccessNode not recycled ${locality}, ${reason}`);
-            })
-            .then(() => {
-              this.transports.delete(transport.id);
-            });
-        }
+        log.debug(`to recycleWorkerNode: ${locality} task:, ${sessionId}`);
+        const taskConfig = {room: this.roomId, task: sessionId};
+        return this.rpcReq.recycleWorkerNode(locality.agent, locality.node, taskConfig)
+          .catch(reason => {
+            log.debug(`AccessNode not recycled ${locality}, ${reason}`);
+          });
       }
     })
     .catch(reason => {

source/portal/versions/requestFormatV1-2.js

@@ -9,6 +9,7 @@
 
 const {
   StreamControlInfo,
+  SubscriptionControlInfo,
 } = require('./requestFormatV1-0');
 
 const Resolution = {
@@ -143,6 +144,7 @@ const SubscriptionRequest = {
 };
 
 // SubscriptionControlInfo
+/*
 const SubscriptionControlInfo = {
   type: 'object',
   anyOf: [
@@ -197,6 +199,7 @@ const SubscriptionControlInfo = {
     }
   }
 };
+*/
 
 module.exports = {
   PublicationRequest,