openvidu-server: skip Kurento remote operations if node crashed

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

@@ -58,6 +58,7 @@ import io.openvidu.server.kurento.endpoint.MediaEndpoint;
 import io.openvidu.server.kurento.endpoint.PublisherEndpoint;
 import io.openvidu.server.kurento.endpoint.SubscriberEndpoint;
 import io.openvidu.server.recording.service.RecordingManager;
+import io.openvidu.server.utils.RemoteOperationUtils;
 
 public class KurentoParticipant extends Participant {
 
@@ -663,7 +664,9 @@ public class KurentoParticipant extends Participant {
 							senderPublisher.numberOfSubscribers--;
 							if (senderPublisher.isPlayerEndpoint() && senderPublisher.numberOfSubscribers == 0) {
 								try {
+									if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 										senderPublisher.getPlayerEndpoint().stop();
+									}
 									log.info(
 											"IP Camera stream {} feed is now disabled because there are no subscribers",
 											senderPublisher.getStreamId());
@@ -691,6 +694,7 @@ public class KurentoParticipant extends Participant {
 	void releaseElement(final String senderName, final MediaElement element) {
 		final String eid = element.getId();
 		try {
+			if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 				element.release(new Continuation<Void>() {
 					@Override
 					public void onSuccess(Void result) throws Exception {
@@ -704,6 +708,7 @@ public class KurentoParticipant extends Participant {
 								eid, senderName, cause);
 					}
 				});
+			}
 		} catch (Exception e) {
 			log.error("PARTICIPANT {}: Error calling release on elem #{} for {}", getParticipantPublicId(), eid,
 					senderName, e);

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

@@ -42,6 +42,7 @@ import io.openvidu.server.core.MediaOptions;
 import io.openvidu.server.core.Participant;
 import io.openvidu.server.core.Session;
 import io.openvidu.server.kurento.kms.Kms;
+import io.openvidu.server.utils.RemoteOperationUtils;
 
 /**
  * @author Pablo Fuente (pablofuenteperez@gmail.com)
@@ -252,6 +253,7 @@ public class KurentoSession extends Session {
 				return;
 			}
 
+			if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 				getPipeline().release(new Continuation<Void>() {
 					@Override
 					public void onSuccess(Void result) throws Exception {
@@ -277,12 +279,13 @@ public class KurentoSession extends Session {
 				});
 			}
 		}
+	}
 
 	public String getParticipantPrivateIdFromStreamId(String streamId) {
 		return this.publishedStreamIds.get(streamId);
 	}
 
-	public void restartStatusInKurento(Long kmsDisconnectionTime) {
+	public void restartStatusInKurentoAfterReconnection(Long kmsDisconnectionTime) {
 
 		log.info("Resetting process: resetting remote media objects for active session {}", this.sessionId);
 

openvidu-server/src/main/java/io/openvidu/server/kurento/endpoint/MediaEndpoint.java

@@ -63,6 +63,7 @@ import io.openvidu.server.config.OpenviduConfig;
 import io.openvidu.server.core.Participant;
 import io.openvidu.server.kurento.core.KurentoMediaOptions;
 import io.openvidu.server.kurento.core.KurentoParticipant;
+import io.openvidu.server.utils.RemoteOperationUtils;
 
 /**
  * {@link Endpoint} wrapper. Can be based on WebRtcEndpoint (that supports
@@ -471,8 +472,10 @@ public abstract class MediaEndpoint {
 		if (element == null || subscription == null) {
 			return;
 		}
+		if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 			element.removeErrorListener(subscription);
 		}
+	}
 
 	/**
 	 * Orders the internal endpoint ({@link RtpEndpoint} or {@link WebRtcEndpoint})

openvidu-server/src/main/java/io/openvidu/server/kurento/endpoint/PublisherEndpoint.java

@@ -52,6 +52,7 @@ import io.openvidu.server.core.MediaOptions;
 import io.openvidu.server.kurento.core.KurentoMediaOptions;
 import io.openvidu.server.kurento.core.KurentoParticipant;
 import io.openvidu.server.utils.JsonUtils;
+import io.openvidu.server.utils.RemoteOperationUtils;
 
 /**
  * Publisher aspect of the {@link MediaEndpoint}.
@@ -333,6 +334,7 @@ public class PublisherEndpoint extends MediaEndpoint {
 		}
 		elementIds.remove(elementId);
 		if (releaseElement) {
+			if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 				element.release(new Continuation<Void>() {
 					@Override
 					public void onSuccess(Void result) throws Exception {
@@ -345,6 +347,7 @@ public class PublisherEndpoint extends MediaEndpoint {
 					}
 				});
 			}
+		}
 		this.filter = null;
 	}
 
@@ -504,6 +507,7 @@ public class PublisherEndpoint extends MediaEndpoint {
 	}
 
 	private void internalSinkDisconnect(final MediaElement source, final MediaElement sink, boolean blocking) {
+		if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 			if (blocking) {
 				source.disconnect(sink);
 			} else {
@@ -522,6 +526,7 @@ public class PublisherEndpoint extends MediaEndpoint {
 				});
 			}
 		}
+	}
 
 	/**
 	 * Same as {@link #internalSinkDisconnect(MediaElement, MediaElement)}, but can
@@ -536,6 +541,7 @@ public class PublisherEndpoint extends MediaEndpoint {
 	 */
 	private void internalSinkDisconnect(final MediaElement source, final MediaElement sink, final MediaType type,
 			boolean blocking) {
+		if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 			if (type == null) {
 				internalSinkDisconnect(source, sink, blocking);
 			} else {
@@ -558,6 +564,7 @@ public class PublisherEndpoint extends MediaEndpoint {
 				}
 			}
 		}
+	}
 
 	public MediaOptions getMediaOptions() {
 		return mediaOptions;

openvidu-server/src/main/java/io/openvidu/server/kurento/kms/KmsManager.java

@@ -51,6 +51,7 @@ import io.openvidu.server.core.SessionManager;
 import io.openvidu.server.kurento.core.KurentoSession;
 import io.openvidu.server.utils.MediaNodeStatusManager;
 import io.openvidu.server.utils.QuarantineKiller;
+import io.openvidu.server.utils.RemoteOperationUtils;
 import io.openvidu.server.utils.UpdatableTimerTask;
 
 public abstract class KmsManager {
@@ -227,7 +228,14 @@ public abstract class KmsManager {
 						log.warn("Closing {} sessions hosted by KMS with uri {}: {}", kms.getKurentoSessions().size(),
 								kms.getUri(), kms.getKurentoSessions().stream().map(s -> s.getSessionId())
 										.collect(Collectors.joining(",", "[", "]")));
+
+						try {
+							// Flag the thread to skip remote operations to KMS
+							RemoteOperationUtils.setToSkipRemoteOperations();
 							sessionManager.closeAllSessionsAndRecordingsOfKms(kms, EndReason.nodeCrashed);
+						} finally {
+							RemoteOperationUtils.revertToRunRemoteOperations();
+						}
 
 						// Remove Media Node
 						log.warn("Removing Media Node {} after crash", kms.getId());
@@ -269,7 +277,7 @@ public abstract class KmsManager {
 										kms.getUri(), kms.getKurentoSessions().size(), kms.getKurentoSessions().stream()
 												.map(s -> s.getSessionId()).collect(Collectors.joining(",", "[", "]")));
 								kms.getKurentoSessions().forEach(kSession -> {
-									kSession.restartStatusInKurento(timeOfKurentoDisconnection);
+									kSession.restartStatusInKurentoAfterReconnection(timeOfKurentoDisconnection);
 								});
 							} else {
 								log.info("KMS with URI {} is the same process. Nothing must be done", kms.getUri());

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

@@ -38,6 +38,7 @@ import io.openvidu.client.OpenViduException;
 import io.openvidu.client.OpenViduException.Code;
 import io.openvidu.server.kurento.core.KurentoSession;
 import io.openvidu.server.kurento.endpoint.PublisherEndpoint;
+import io.openvidu.server.utils.RemoteOperationUtils;
 
 public class CompositeWrapper {
 
@@ -87,7 +88,21 @@ public class CompositeWrapper {
 	}
 
 	public synchronized void stopCompositeRecording(CountDownLatch stopLatch, Long kmsDisconnectionTime) {
-		if (kmsDisconnectionTime == null) {
+
+		if (kmsDisconnectionTime != null || RemoteOperationUtils.mustSkipRemoteOperation()) {
+			// Stopping composite endpoint because of a KMS disconnection
+			String msg;
+			if (kmsDisconnectionTime != null) {
+				endTime = kmsDisconnectionTime;
+				msg = "KMS restart";
+			} else {
+				endTime = System.currentTimeMillis();
+				msg = "node crashed";
+			}
+			stopLatch.countDown();
+			log.warn("Forcing composed audio-only recording stop after {} in session {}", msg,
+					this.session.getSessionId());
+		} else {
 			this.recorderEndpoint.addStoppedListener(new EventListener<StoppedEvent>() {
 				@Override
 				public void onEvent(StoppedEvent event) {
@@ -100,11 +115,6 @@ public class CompositeWrapper {
 				}
 			});
 			this.recorderEndpoint.stop();
-		} else {
-			endTime = kmsDisconnectionTime;
-			stopLatch.countDown();
-			log.warn("Forcing composed audio-only recording stop after KMS restart in session {}",
-					this.session.getSessionId());
 		}
 
 	}
@@ -145,7 +155,9 @@ public class CompositeWrapper {
 		HubPort hubPort = this.hubPorts.remove(streamId);
 		PublisherEndpoint publisherEndpoint = this.publisherEndpoints.remove(streamId);
 		publisherEndpoint.disconnectFrom(hubPort);
+		if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 			hubPort.release();
+		}
 		log.info("Composite for session {} has now {} connected publishers", this.session.getSessionId(),
 				this.composite.getChildren().size() - 1);
 	}
@@ -155,12 +167,16 @@ public class CompositeWrapper {
 			PublisherEndpoint endpoint = this.publisherEndpoints.get(streamId);
 			HubPort hubPort = this.hubPorts.get(streamId);
 			endpoint.disconnectFrom(hubPort);
+			if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 				hubPort.release();
+			}
 		});
 		this.hubPorts.clear();
 		this.publisherEndpoints.clear();
+		if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 			this.composite.release();
 		}
+	}
 
 	public long getDuration() {
 		return this.endTime - this.startTime;

openvidu-server/src/main/java/io/openvidu/server/recording/service/RecordingManager.java

@@ -31,8 +31,9 @@ import java.util.NoSuchElementException;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
@@ -79,6 +80,7 @@ import io.openvidu.server.utils.JsonUtils;
 import io.openvidu.server.utils.LocalCustomFileManager;
 import io.openvidu.server.utils.LocalDockerManager;
 import io.openvidu.server.utils.RecordingUtils;
+import io.openvidu.server.utils.RemoteOperationUtils;
 
 public class RecordingManager {
 
@@ -122,8 +124,8 @@ public class RecordingManager {
 
 	private JsonUtils jsonUtils = new JsonUtils();
 
-	private ScheduledThreadPoolExecutor automaticRecordingStopExecutor = new ScheduledThreadPoolExecutor(
+	private ScheduledExecutorService automaticRecordingStopExecutor = Executors
-			Runtime.getRuntime().availableProcessors());
+			.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());
 
 	public static final String IMAGE_NAME = "openvidu/openvidu-recording";
 
@@ -809,9 +811,11 @@ public class RecordingManager {
 					throw new OpenViduException(Code.RECORDING_PATH_NOT_VALID, errorMessage);
 				}
 
+				if (!RemoteOperationUtils.mustSkipRemoteOperation()) {
 					recorder.stop();
 					recorder.release();
 					pipeline.release();
+				}
 
 				log.info("Kurento Media Server has write permissions on recording path: {}", openviduRecordingPath);
 

openvidu-server/src/main/java/io/openvidu/server/recording/service/SingleStreamRecordingService.java

@@ -68,6 +68,7 @@ import io.openvidu.server.recording.Recording;
 import io.openvidu.server.recording.RecordingDownloader;
 import io.openvidu.server.recording.RecordingUploader;
 import io.openvidu.server.utils.CustomFileManager;
+import io.openvidu.server.utils.RemoteOperationUtils;
 
 public class SingleStreamRecordingService extends RecordingService {
 
@@ -289,7 +290,20 @@ public class SingleStreamRecordingService extends RecordingService {
 			try {
 				if (kParticipant.singleRecordingLock.tryLock(15, TimeUnit.SECONDS)) {
 					try {
-						if (kmsDisconnectionTime == null) {
+
+						if (kmsDisconnectionTime != null || RemoteOperationUtils.mustSkipRemoteOperation()) {
+
+							// Stopping recorder endpoint because of a KMS disconnection
+							finalWrapper.setEndTime(
+									kmsDisconnectionTime != null ? kmsDisconnectionTime : System.currentTimeMillis());
+							generateIndividualMetadataFile(finalWrapper);
+							globalStopLatch.countDown();
+							log.warn("Forcing individual recording stop after {} for stream {} in recording {}",
+									kmsDisconnectionTime != null ? "KMS restart" : "node crashed", streamId,
+									recordingId);
+
+						} else {
+
 							finalWrapper.getRecorder().addStoppedListener(new EventListener<StoppedEvent>() {
 								@Override
 								public void onEvent(StoppedEvent event) {
@@ -301,14 +315,7 @@ public class SingleStreamRecordingService extends RecordingService {
 								}
 							});
 							finalWrapper.getRecorder().stop();
-						} else {
+
-							// Stopping recorder endpoint because of a KMS disconnection
-							finalWrapper.setEndTime(kmsDisconnectionTime);
-							generateIndividualMetadataFile(finalWrapper);
-							globalStopLatch.countDown();
-							log.warn(
-									"Forcing individual recording stop after KMS restart for stream {} in recording {}",
-									streamId, recordingId);
 						}
 					} finally {
 						kParticipant.singleRecordingLock.unlock();

openvidu-server/src/main/java/io/openvidu/server/utils/RemoteOperationUtils.java

@@ -0,0 +1,21 @@
+package io.openvidu.server.utils;
+
+public class RemoteOperationUtils {
+
+	private final static String VALUE = "SKIP_REMOTE_OPERATION";
+
+	private static final ThreadLocal<String> threadLocal = ThreadLocal.withInitial(() -> "");
+
+	public static void setToSkipRemoteOperations() {
+		threadLocal.set(VALUE);
+	}
+
+	public static boolean mustSkipRemoteOperation() {
+		return VALUE.equals(threadLocal.get());
+	}
+
+	public static void revertToRunRemoteOperations() {
+		threadLocal.remove();
+	}
+
+}