openvidu-server: automatically stip MANUAL recording after timeout

openvidu-server/src/main/java/io/openvidu/server/kurento/core/KurentoSessionManager.java

@@ -175,14 +175,21 @@ public class KurentoSessionManager extends SessionManager {
 			showTokens();
 		} else if (remainingParticipants.size() == 1 && openviduConfig.isRecordingModuleEnabled()
 				&& MediaMode.ROUTED.equals(session.getSessionProperties().mediaMode())
-				&& RecordingMode.ALWAYS.equals(session.getSessionProperties().recordingMode())
 				&& ProtocolElements.RECORDER_PARTICIPANT_PUBLICID
 						.equals(remainingParticipants.iterator().next().getParticipantPublicId())) {
-
+			if (RecordingMode.ALWAYS.equals(session.getSessionProperties().recordingMode())) {
+				// Immediately stop recording when last real participant left if
+				// RecordingMode.ALWAYS
 				log.info("Last participant left. Stopping recording for session {}", sessionId);
 				recordingService.stopRecording(session, null, reason);
 				evictParticipant(session.getParticipantByPublicId(ProtocolElements.RECORDER_PARTICIPANT_PUBLICID), null,
 						null, "EVICT_RECORDER");
+			} else if (RecordingMode.MANUAL.equals(session.getSessionProperties().recordingMode())) {
+				// Start countdown to stop recording if RecordingMode.MANUAL (will be aborted if
+				// a Publisher starts before timeout)
+				log.info("Last participant left. Starting countdown for stopping recording of session {}", sessionId);
+				recordingService.initAutomaticRecordingStopThread(session.getSessionId());
+			}
 		}
 
 		// Finally close websocket session if required
@@ -206,14 +213,10 @@ public class KurentoSessionManager extends SessionManager {
 	 * the peer's request by sending it the SDP response (answer or updated offer)
 	 * generated by the WebRTC endpoint on the server.
 	 *
-	 * @param participant
+	 * @param participant   Participant publishing video
-	 *            Participant publishing video
+	 * @param MediaOptions  configuration of the stream to publish
-	 * @param MediaOptions
+	 * @param transactionId identifier of the Transaction
-	 *            configuration of the stream to publish
+	 * @throws OpenViduException on error
-	 * @param transactionId
-	 *            identifier of the Transaction
-	 * @throws OpenViduException
-	 *             on error
 	 */
 	@Override
 	public void publishVideo(Participant participant, MediaOptions mediaOptions, Integer transactionId)
@@ -272,9 +275,9 @@ public class KurentoSessionManager extends SessionManager {
 
 		if (this.openviduConfig.isRecordingModuleEnabled()
 				&& MediaMode.ROUTED.equals(session.getSessionProperties().mediaMode())
-				&& RecordingMode.ALWAYS.equals(session.getSessionProperties().recordingMode())
-				&& !recordingService.sessionIsBeingRecorded(session.getSessionId())
 				&& session.getActivePublishers() == 0) {
+			if (RecordingMode.ALWAYS.equals(session.getSessionProperties().recordingMode())
+					&& !recordingService.sessionIsBeingRecorded(session.getSessionId())) {
 				// Insecure session recording
 				new Thread(() -> {
 					recordingService.startRecording(session,
@@ -282,6 +285,19 @@ public class KurentoSessionManager extends SessionManager {
 									.recordingLayout(session.getSessionProperties().defaultRecordingLayout())
 									.customLayout(session.getSessionProperties().defaultCustomLayout()).build());
 				}).start();
+			} else if (RecordingMode.MANUAL.equals(session.getSessionProperties().recordingMode())
+					&& recordingService.sessionIsBeingRecorded(session.getSessionId())) {
+				// Abort automatic recording stop (user published before timeout)
+				log.info("Participant {} published before timeout finished. Aborting automatic recording stop",
+						participant.getParticipantPublicId());
+				boolean stopAborted = recordingService.abortAutomaticRecordingStopThread(session.getSessionId());
+				if (stopAborted) {
+					log.info("Automatic recording stopped succesfully aborted");
+				} else {
+					log.info("Automatic recording stopped couldn't be aborted. Recording of session {} has stopped",
+							session.getSessionId());
+				}
+			}
 		}
 
 		session.newPublisher(participant);
@@ -458,12 +474,10 @@ public class KurentoSessionManager extends SessionManager {
 	 * Creates a session if it doesn't already exist. The session's id will be
 	 * indicated by the session info bean.
 	 *
-	 * @param kcSessionInfo
+	 * @param kcSessionInfo bean that will be passed to the
-	 *            bean that will be passed to the {@link KurentoClientProvider} in
+	 *                      {@link KurentoClientProvider} in order to obtain the
-	 *            order to obtain the {@link KurentoClient} that will be used by the
+	 *                      {@link KurentoClient} that will be used by the room
-	 *            room
+	 * @throws OpenViduException in case of error while creating the session
-	 * @throws OpenViduException
-	 *             in case of error while creating the session
 	 */
 	public void createSession(KurentoClientSessionInfo kcSessionInfo, SessionProperties sessionProperties)
 			throws OpenViduException {

openvidu-server/src/main/java/io/openvidu/server/recording/ComposedRecordingService.java

@@ -31,6 +31,8 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -87,6 +89,10 @@ public class ComposedRecordingService {
 	private Map<String, Recording> startingRecordings = new ConcurrentHashMap<>();
 	private Map<String, Recording> startedRecordings = new ConcurrentHashMap<>();
 	private Map<String, Recording> sessionsRecordings = new ConcurrentHashMap<>();
+	private final Map<String, ScheduledFuture<?>> automaticRecordingStopThreads = new ConcurrentHashMap<>();
+
+	private ScheduledThreadPoolExecutor automaticRecordingStopExecutor = new ScheduledThreadPoolExecutor(
+			Runtime.getRuntime().availableProcessors());
 
 	private final String IMAGE_NAME = "openvidu/openvidu-recording";
 	private String IMAGE_TAG;
@@ -162,8 +168,8 @@ public class ComposedRecordingService {
 		if (session == null) {
 			log.warn(
 					"Existing recording {} does not have an active session associated. This usually means the recording"
-							+ " layout did not join a recorded participant and therefore the session closed before"
+							+ " layout did not join a recorded participant or the recording has been automatically"
-							+ " stopping the recording container",
+							+ " stopped after last user left and timeout passed",
 					recordingId);
 			recording = this.startedRecordings.remove(recordingId);
 			containerId = this.sessionsContainers.remove(recording.getSessionId());
@@ -543,4 +549,19 @@ public class ComposedRecordingService {
 		this.IMAGE_TAG = version;
 	}
 
+	public void initAutomaticRecordingStopThread(String sessionId) {
+		final String recordingId = this.sessionsRecordings.get(sessionId).getId();
+		ScheduledFuture<?> future = this.automaticRecordingStopExecutor.schedule(() -> {
+			log.info("Stopping recording {} after 2 minutes wait (no publisher published before timeout)", recordingId);
+			this.stopRecording(null, recordingId, "lastParticipantLeft");
+			this.automaticRecordingStopThreads.remove(sessionId);
+		}, 2, TimeUnit.MINUTES);
+		this.automaticRecordingStopThreads.putIfAbsent(sessionId, future);
+	}
+
+	public boolean abortAutomaticRecordingStopThread(String sessionId) {
+		ScheduledFuture<?> future = this.automaticRecordingStopThreads.remove(sessionId);
+		return future.cancel(false);
+	}
+
 }