handle audio only participants in a room

FilterPlugin/src/main/java/io/antmedia/filter/utils/MCUFilterTextGenerator.java

@@ -1,36 +1,47 @@
 package io.antmedia.filter.utils;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.antmedia.plugin.MCUManager;
 
 public class MCUFilterTextGenerator {
-	
+
 	private static Logger logger = LoggerFactory.getLogger(MCUFilterTextGenerator.class);
-	
+
 	public static String createAudioFilter(int streamCount) {
 		if(streamCount == 1) {
 			return "[in0]acopy[out0]";
 		}
-		
+
 		String filter = "";
-        for (int i = 0; i < streamCount; i++) {
-            filter += "[in" + i + "]";
-        }
-        filter += "amix=inputs=" + streamCount + "[out0]";
+		for (int i = 0; i < streamCount; i++) {
+			filter += "[in" + i + "]";
+		}
+		filter += "amix=inputs=" + streamCount + "[out0]";
 
-        return filter;
+		return filter;
 	}
 
 	public static String createVideoFilter(int streamCount) {
-
+		List<Integer> inputIndices = new ArrayList<Integer>();
+		for (int i = 0; i < streamCount; i++) {
+			inputIndices.add(i);
+		}
+		return createVideoFilter(streamCount, inputIndices);
+	}
+
+
+	public static String createVideoFilter(int streamCount, List<Integer> inputIndices) {
 		int width = 360;
 		int height = 240;
 		String color = "black";
 		int margin = 3;
-		
+
 		if(streamCount == 1) {
 			return "[in0]copy[out0]";
 		}
@@ -39,12 +50,12 @@ public static String createVideoFilter(int streamCount) {
 		int columns = (int) Math.ceil(Math.sqrt((double)streamCount));
 		int rows = (int) Math.ceil((double)streamCount/columns);
 		int lastRowColumns = streamCount - (rows - 1) * columns;
-		
+
 		width = Math.min(360, 720/columns);
 		height = 240*width/360;
 
 		for (int i = 0; i < streamCount; i++) {
-			filter += "[in" + i + "]scale="+(width-2*margin)+":"+(height-2*margin)+":force_original_aspect_ratio=decrease";
+			filter += "[in" + inputIndices.get(i) + "]scale="+(width-2*margin)+":"+(height-2*margin)+":force_original_aspect_ratio=decrease";
 			filter += ",pad="+width+":"+height+":"+margin+":"+margin+":color="+color;
 			filter += "[s" + i + "];";
 		}
@@ -70,9 +81,11 @@ public static String createVideoFilter(int streamCount) {
 			}
 			filter += "vstack=inputs=" + rows + ",pad=720:480:(ow-iw)/2:(oh-ih)/2[out0]";
 		}
-		
+
 		logger.info("generated filter:{}", filter);
 
 		return filter;
 	}
+
+
 }

FilterPlugin/src/main/java/io/antmedia/plugin/MCUManager.java

@@ -6,7 +6,6 @@
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
-import org.apache.commons.lang3.RandomUtils;
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -23,6 +22,8 @@
 import io.antmedia.filter.utils.MCUFilterTextGenerator;
 import io.antmedia.muxer.IAntMediaStreamHandler;
 import io.antmedia.plugin.api.IStreamListener;
+import io.antmedia.webrtc.VideoCodec;
+import io.antmedia.webrtc.api.IWebRTCAdaptor;
 import io.antmedia.websocket.WebSocketConstants;
 
 @Component(value="filters.mcu")
@@ -40,6 +41,8 @@ public class MCUManager implements ApplicationContextAware, IStreamListener{
 	private static Logger logger = LoggerFactory.getLogger(MCUManager.class);
 	private Queue<String> roomsHasCustomFilters = new ConcurrentLinkedQueue<>();
 	private Queue<String> customRooms = new ConcurrentLinkedQueue<>();
+
+	private IWebRTCAdaptor webRTCAdaptor = null;
 
 
 
@@ -77,6 +80,13 @@ public AntMediaApplicationAdapter getApplication() {
 		}
 		return appAdaptor;
 	}
+
+	public IWebRTCAdaptor getWebRTCAdaptor() {
+		if(webRTCAdaptor == null) {
+			webRTCAdaptor = (IWebRTCAdaptor) applicationContext.getBean(IWebRTCAdaptor.BEAN_NAME);
+		}
+		return webRTCAdaptor;
+	}
 
 	public FiltersManager getFiltersManager() {
 		if(filtersManager == null) {
@@ -99,13 +109,23 @@ else if (!roomsHasCustomFilters.contains(roomId))
 			//Update room filter if there is no custom filter
 			try {
 				List<String> streams = new ArrayList<>();
+				List<Integer> videoEnabledIndices = new ArrayList<>();
+
 				streams.addAll(room.getRoomStreamList());
 
+				int index = 0;
 				for (String streamId : room.getRoomStreamList()) {
 					Broadcast broadcast = datastore.get(streamId);
 					if(broadcast == null || !broadcast.getStatus().equals(IAntMediaStreamHandler.BROADCAST_STATUS_BROADCASTING)) {
 						streams.remove(streamId);
 					}
+					else {
+						boolean isVideoEnabled = isVideoEnabled(streamId);
+						if(isVideoEnabled) {
+							videoEnabledIndices.add(index);	
+						}
+						index++;
+					}
 				}
 				//
 				if (!streams.isEmpty()) 
@@ -116,7 +136,7 @@ else if (!roomsHasCustomFilters.contains(roomId))
 					List<String> outputStreams = new ArrayList<>();
 					outputStreams.add(roomId+MERGED_SUFFIX);
 					filterConfiguration.setOutputStreams(outputStreams);
-					filterConfiguration.setVideoFilter(MCUFilterTextGenerator.createVideoFilter(streams.size()));
+					filterConfiguration.setVideoFilter(MCUFilterTextGenerator.createVideoFilter(videoEnabledIndices.size(), videoEnabledIndices));
 					filterConfiguration.setAudioFilter(MCUFilterTextGenerator.createAudioFilter(streams.size()));
 					filterConfiguration.setVideoEnabled(!room.getMode().equals(WebSocketConstants.AMCU));
 					filterConfiguration.setAudioEnabled(true);
@@ -154,6 +174,10 @@ else if(!room.isZombi()) {
 		return result;
 	}
 
+	public boolean isVideoEnabled(String streamId) {
+		return getWebRTCAdaptor().getStreamInfo(streamId).get(0).getVideoCodec() != VideoCodec.NOVIDEO;
+	}
+
 	private void roomHasChange(String roomId) {
 		DataStore datastore = getApplication().getDataStore();
 		ConferenceRoom room = datastore.getConferenceRoom(roomId);	

FilterPlugin/src/test/java/io/antmedia/test/MCUManagerUnitTest.java

@@ -1,6 +1,10 @@
 package io.antmedia.test;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doNothing;
@@ -13,7 +17,10 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
 
 import org.apache.commons.lang3.RandomUtils;
 import org.apache.commons.lang3.exception.ExceptionUtils;
@@ -22,16 +29,25 @@
 import org.junit.rules.TestRule;
 import org.junit.rules.TestWatcher;
 import org.junit.runner.Description;
+import org.mockito.ArgumentCaptor;
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
 
 import io.antmedia.AntMediaApplicationAdapter;
+import io.antmedia.cluster.IStreamInfo;
 import io.antmedia.datastore.db.DataStore;
 import io.antmedia.datastore.db.InMemoryDataStore;
 import io.antmedia.datastore.db.types.Broadcast;
 import io.antmedia.datastore.db.types.ConferenceRoom;
+import io.antmedia.datastore.db.types.StreamInfo;
+import io.antmedia.filter.utils.FilterConfiguration;
+import io.antmedia.filter.utils.MCUFilterTextGenerator;
 import io.antmedia.muxer.IAntMediaStreamHandler;
 import io.antmedia.plugin.FiltersManager;
 import io.antmedia.plugin.MCUManager;
 import io.antmedia.rest.model.Result;
+import io.antmedia.webrtc.VideoCodec;
+import io.antmedia.webrtc.api.IWebRTCAdaptor;
 import io.antmedia.websocket.WebSocketConstants;
 import io.vertx.core.Vertx;
 
@@ -113,9 +129,11 @@ public void testMCUWithOtherRooms() {
 		String roomId = "room"+RandomUtils.nextInt();
 		MCUManager mcuManager = spy(new MCUManager());
 		FiltersManager filtersManager = spy(new FiltersManager());
+		IWebRTCAdaptor webRTCAdaptor = mock(IWebRTCAdaptor.class);
 
 		doReturn(filtersManager).when(mcuManager).getFiltersManager();
-
+		doReturn(true).when(mcuManager).isVideoEnabled(anyString());
+
 		doReturn(new Result(true)).when(filtersManager).createFilter(any(), any());
 
 
@@ -149,4 +167,130 @@ public void testMCUWithOtherRooms() {
 
 	}
 
+	/*
+	 * This test tests to generated filter text for the rooms who has both audio only and normal participants
+	 */
+	@Test
+	public void testFilterTextForMixedRoom() {
+
+		ApplicationContext applicationContext = mock(ApplicationContext.class);
+
+		String roomId = "room"+RandomUtils.nextInt();
+		MCUManager mcuManager = spy(new MCUManager());
+		FiltersManager filtersManager = mock(FiltersManager.class);
+		doReturn(filtersManager).when(mcuManager).getFiltersManager();
+		Result result = new Result(true);
+		when(filtersManager.createFilter(any(), any())).thenReturn(result );
+
+		AntMediaApplicationAdapter app = mock(AntMediaApplicationAdapter.class);
+		when(applicationContext.getBean(AntMediaApplicationAdapter.BEAN_NAME)).thenReturn(app);
+
+		DataStore dataStore = mock(DataStore.class);
+		when(app.getDataStore()).thenReturn(dataStore);
+		doNothing().when(app).addStreamListener(mcuManager);
+
+		doReturn(app).when(mcuManager).getApplication();
+		Vertx vertx = mock(Vertx.class);
+		when(vertx.setPeriodic(anyLong(), any())).thenReturn(5l);
+
+		when(app.getVertx()).thenReturn(vertx );
+
+		IWebRTCAdaptor webRTCAdaptor = mock(IWebRTCAdaptor.class);
+		when(applicationContext.getBean(IWebRTCAdaptor.BEAN_NAME)).thenReturn(webRTCAdaptor);
+
+		mcuManager.setApplicationContext(applicationContext);
+
+
+		String s1 = "stream1";
+		List<IStreamInfo> siList1 = new ArrayList<IStreamInfo>();
+		IStreamInfo si1 = mock(IStreamInfo.class);
+		when(si1.getVideoCodec()).thenReturn(VideoCodec.H264);
+		siList1.add(si1);
+		Broadcast broadcast1 = new Broadcast(IAntMediaStreamHandler.BROADCAST_STATUS_BROADCASTING, s1);
+		try {
+			broadcast1.setStreamId(s1);
+		} catch (Exception e) {
+			e.printStackTrace();
+		}
+		when(dataStore.get(s1)).thenReturn(broadcast1);
+
+
+
+		String s2 = "stream2";
+		List<IStreamInfo> siList2 = new ArrayList<IStreamInfo>();
+		IStreamInfo si2 = mock(IStreamInfo.class);
+		when(si2.getVideoCodec()).thenReturn(VideoCodec.NOVIDEO);
+		siList2.add(si2);
+		Broadcast broadcast2 = new Broadcast(IAntMediaStreamHandler.BROADCAST_STATUS_BROADCASTING, s2);
+		try {
+			broadcast1.setStreamId(s2);
+		} catch (Exception e) {
+			e.printStackTrace();
+		}
+		when(dataStore.get(s2)).thenReturn(broadcast2);
+
+		String s3 = "stream3";
+		List<IStreamInfo> siList3 = new ArrayList<IStreamInfo>();
+		IStreamInfo si3 = mock(IStreamInfo.class);
+		when(si3.getVideoCodec()).thenReturn(VideoCodec.H264);
+		siList3.add(si3);
+		Broadcast broadcast3 = new Broadcast(IAntMediaStreamHandler.BROADCAST_STATUS_BROADCASTING, s3);
+		try {
+			broadcast3.setStreamId(s3);
+		} catch (Exception e) {
+			e.printStackTrace();
+		}
+		when(dataStore.get(s3)).thenReturn(broadcast3);
+
+		when(webRTCAdaptor.getStreamInfo(s1)).thenReturn(siList1);
+		when(webRTCAdaptor.getStreamInfo(s2)).thenReturn(siList2);
+		when(webRTCAdaptor.getStreamInfo(s3)).thenReturn(siList3);
+
+
+		ConferenceRoom room = new ConferenceRoom();
+		room.setMode(WebSocketConstants.MCU);
+		room.setRoomId(roomId);
+		when(dataStore.getConferenceRoom(roomId)).thenReturn(room);
+		ArgumentCaptor<FilterConfiguration> filterConf = ArgumentCaptor.forClass(FilterConfiguration.class);
+
+		room.getRoomStreamList().add(s1);
+		mcuManager.updateRoomFilter(roomId);
+		verify(filtersManager, times(1)).createFilter(filterConf.capture(), eq(app));
+		assertTrue(filterConf.getValue().getVideoFilter().contains("in0"));
+
+		room.getRoomStreamList().add(s2);
+		mcuManager.updateRoomFilter(roomId);
+		verify(filtersManager, times(2)).createFilter(filterConf.capture(), eq(app));
+		assertTrue(filterConf.getValue().getVideoFilter().contains("in0"));
+		assertFalse(filterConf.getValue().getVideoFilter().contains("in1"));
+
+		room.getRoomStreamList().add(s3);
+		mcuManager.updateRoomFilter(roomId);
+		verify(filtersManager, times(3)).createFilter(filterConf.capture(), eq(app));
+		assertTrue(filterConf.getValue().getVideoFilter().contains("in0"));
+		assertFalse(filterConf.getValue().getVideoFilter().contains("in1"));
+		assertTrue(filterConf.getValue().getVideoFilter().contains("in2"));
+
+	}
+
+	@Test
+	public void testCreateVideoFilter() {
+		String filter = MCUFilterTextGenerator.createVideoFilter(5);
+		assertTrue(filter.contains("in0"));
+		assertTrue(filter.contains("in1"));
+		assertTrue(filter.contains("in2"));
+		assertTrue(filter.contains("in3"));
+		assertTrue(filter.contains("in4"));
+
+
+		List<Integer> inputIndices = new ArrayList<Integer>();
+		inputIndices.add(1);
+		inputIndices.add(3);
+		String filter2 = MCUFilterTextGenerator.createVideoFilter(inputIndices.size(), inputIndices);
+		assertTrue(!filter2.contains("in0"));
+		assertTrue(filter2.contains("in1"));
+		assertTrue(!filter2.contains("in2"));
+		assertTrue(filter2.contains("in3"));
+	}
+
 }