Merge branch 'master' into fix-leaked-socket

Author: burak-58

src/main/java/io/antmedia/streamsource/StreamFetcherManager.java

@@ -540,23 +540,20 @@ public void controlStreamFetchers(boolean restart) {
 
 			//get the updated broadcast object
 			Broadcast broadcast = datastore.get(streamScheduler.getStreamId());
-			
-			if  (broadcast != null && AntMediaApplicationAdapter.PLAY_LIST.equals(broadcast.getType())) {
-				//if it's playlist, continue
+
+			if (broadcast != null && AntMediaApplicationAdapter.PLAY_LIST.equals(broadcast.getType())) {
+				// For playlists, only check auto-stop if autoStartStopEnabled is true
+				if (broadcast.isAutoStartStopEnabled() && isToBeStoppedAutomatically(broadcast)) {
+					logger.info("Auto-stopping playlist {} because no viewers are watching", streamScheduler.getStreamId());
+					stopPlayList(streamScheduler.getStreamId());
+				}
 				continue;
 			}
-			
+
 			boolean autoStop = false;
-			if (restart || broadcast == null || 
-					(autoStop = isToBeStoppedAutomatically(broadcast))) 
+			if (restart || broadcast == null ||
+					(autoStop = isToBeStoppedAutomatically(broadcast)))
 			{
-				//logic of If condition is
-				
-				// stop it if it's restart = true 
-				//   or 
-				// brodcast == null because it means stream is deleted
-				//  or
-				// autoStop
 
 				logger.info("Calling stop stream {} due to restart -> {}, broadcast is null -> {}, auto stop because no viewer -> {}", 
 						streamScheduler.getStreamId(), restart, broadcast == null, autoStop);

src/main/java/io/antmedia/websocket/WebSocketCommunityHandler.java

@@ -153,7 +153,7 @@ else if (cmd.equals(WebSocketConstants.STOP_COMMAND)) {
 			else if (cmd.equals(WebSocketConstants.PING_COMMAND)) {
 				sendPongMessage(session);
 			}
-			else if (cmd.equals(WebSocketConstants.GET_STREAM_INFO_COMMAND) || cmd.equals(WebSocketConstants.PLAY_COMMAND)) 
+			else if (cmd.equals(WebSocketConstants.GET_STREAM_INFO_COMMAND) || cmd.equals(WebSocketConstants.PLAY_COMMAND))
 			{
 				sendNotFoundJSON(streamId, session);
 			}
@@ -166,7 +166,6 @@ else if (cmd.equals(WebSocketConstants.GET_STREAM_INFO_COMMAND) || cmd.equals(We
 		}
 
 	}
-		
 
 	private void startRTMPAdaptor(Session session, final String streamId, boolean enableVideo) {
 		int rtmpPort = appAdaptor.getServerSettings().getRtmpPort();

src/test/java/io/antmedia/test/StreamSchedularUnitTest.java

@@ -732,7 +732,7 @@ public void testIsStreamRunning()
 
 	@Test
 	public void testControlStreamFetchersPlayListAndRestart() {
-		DataStore dataStore = Mockito.mock(DataStore.class); 
+		DataStore dataStore = Mockito.mock(DataStore.class);
 		StreamFetcherManager streamFetcherManager = Mockito.spy(new StreamFetcherManager(vertx, dataStore, appScope));
 		Map<String, StreamFetcher> streamFetcherList = new ConcurrentHashMap<>();
 
@@ -742,7 +742,7 @@ public void testControlStreamFetchersPlayListAndRestart() {
 		streamFetcherList.put(streamId, fetcher);
 		Mockito.when(fetcher.getStreamId()).thenReturn(streamId);
 		Mockito.when(fetcher.getStreamUrl()).thenReturn(streamUrl);
-		
+
 		when(fetcher.isStreamAlive()).thenReturn(true);
 		when(fetcher.isStreamBlocked()).thenReturn(false);
 
@@ -751,11 +751,12 @@ public void testControlStreamFetchersPlayListAndRestart() {
 		when(broadcast.getStreamId()).thenReturn(streamId);
 		when(broadcast.getStreamUrl()).thenReturn("streamurl");
 		when(broadcast.getType()).thenReturn(AntMediaApplicationAdapter.PLAY_LIST);
+		when(broadcast.isAutoStartStopEnabled()).thenReturn(false);
 
 		streamFetcherManager.setStreamFetcherList(streamFetcherList);
 
 		streamFetcherManager.controlStreamFetchers(false);
-		//it should not call anything because type is playlist
+		//it should not call isToBeStoppedAutomatically because type is playlist and autoStartStopEnabled is false
 		Mockito.verify(streamFetcherManager, Mockito.never()).isToBeStoppedAutomatically(Mockito.any());
 
 
@@ -773,6 +774,47 @@ public void testControlStreamFetchersPlayListAndRestart() {
 		Mockito.verify(streamFetcherManager).startStreaming(broadcast);
 
 	}
+
+	@Test
+	public void testControlStreamFetchersPlayListAutoStop() {
+		DataStore dataStore = Mockito.mock(DataStore.class);
+		StreamFetcherManager streamFetcherManager = Mockito.spy(new StreamFetcherManager(vertx, dataStore, appScope));
+		Map<String, StreamFetcher> streamFetcherList = new ConcurrentHashMap<>();
+
+		StreamFetcher fetcher = Mockito.mock(StreamFetcher.class);
+		String streamId = "playlistStream123";
+		String streamUrl = "streamurl";
+		streamFetcherList.put(streamId, fetcher);
+		Mockito.when(fetcher.getStreamId()).thenReturn(streamId);
+		Mockito.when(fetcher.getStreamUrl()).thenReturn(streamUrl);
+
+		when(fetcher.isStreamAlive()).thenReturn(true);
+		when(fetcher.isStreamBlocked()).thenReturn(false);
+
+		Broadcast broadcast = mock(Broadcast.class);
+		when(dataStore.get(Mockito.any())).thenReturn(broadcast);
+		when(broadcast.getStreamId()).thenReturn(streamId);
+		when(broadcast.getStreamUrl()).thenReturn("streamurl");
+		when(broadcast.getType()).thenReturn(AntMediaApplicationAdapter.PLAY_LIST);
+		when(broadcast.isAutoStartStopEnabled()).thenReturn(true);
+
+		streamFetcherManager.setStreamFetcherList(streamFetcherList);
+
+		// When autoStartStopEnabled is true, isToBeStoppedAutomatically should be called for playlists
+		Mockito.doReturn(false).when(streamFetcherManager).isToBeStoppedAutomatically(Mockito.any());
+		streamFetcherManager.controlStreamFetchers(false);
+		Mockito.verify(streamFetcherManager, Mockito.times(1)).isToBeStoppedAutomatically(broadcast);
+
+		// Reset and test when isToBeStoppedAutomatically returns true - should call stopPlayList
+		Mockito.reset(streamFetcherManager);
+		streamFetcherManager.setStreamFetcherList(streamFetcherList);
+		Mockito.doReturn(true).when(streamFetcherManager).isToBeStoppedAutomatically(Mockito.any());
+		Mockito.doReturn(new Result(true)).when(streamFetcherManager).stopPlayList(Mockito.any());
+
+		streamFetcherManager.controlStreamFetchers(false);
+		Mockito.verify(streamFetcherManager, Mockito.times(1)).isToBeStoppedAutomatically(broadcast);
+		Mockito.verify(streamFetcherManager, Mockito.times(1)).stopPlayList(streamId);
+	}
 
 
 	@Test