You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@openmeetings.apache.org by so...@apache.org on 2019/12/18 11:44:34 UTC
[openmeetings] branch master updated: [OPENMEETINGS-2153] media
stream is dropped with configurable time-out
This is an automated email from the ASF dual-hosted git repository.
solomax pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/openmeetings.git
The following commit(s) were added to refs/heads/master by this push:
new bf6dc2d [OPENMEETINGS-2153] media stream is dropped with configurable time-out
bf6dc2d is described below
commit bf6dc2dca5af6ebc4a3150dc61eb89ba2120835e
Author: Maxim Solodovnik <so...@gmail.com>
AuthorDate: Wed Dec 18 18:44:19 2019 +0700
[OPENMEETINGS-2153] media stream is dropped with configurable time-out
---
.../apache/openmeetings/core/remote/KStream.java | 35 +++++++++++++++++-----
.../openmeetings/core/remote/KurentoHandler.java | 15 ++++++++--
.../webapp/WEB-INF/classes/applicationContext.xml | 1 +
3 files changed, 40 insertions(+), 11 deletions(-)
diff --git a/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/KStream.java b/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/KStream.java
index fa5b917..e2124d4 100644
--- a/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/KStream.java
+++ b/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/KStream.java
@@ -22,15 +22,20 @@
package org.apache.openmeetings.core.remote;
import static java.util.UUID.randomUUID;
+import static java.util.concurrent.CompletableFuture.delayedExecutor;
import static org.apache.openmeetings.core.remote.KurentoHandler.PARAM_CANDIDATE;
import static org.apache.openmeetings.core.remote.KurentoHandler.PARAM_ICE;
+import static org.apache.openmeetings.core.remote.KurentoHandler.getFlowoutTimeout;
import static org.apache.openmeetings.core.remote.KurentoHandler.newKurentoMsg;
import static org.apache.openmeetings.util.OmFileHelper.getRecUri;
import static org.apache.openmeetings.util.OmFileHelper.getRecordingChunk;
import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
import org.apache.openmeetings.core.util.WebSocketHelper;
import org.apache.openmeetings.db.entity.basic.Client;
@@ -42,7 +47,6 @@ import org.apache.openmeetings.db.util.ws.RoomMessage;
import org.apache.openmeetings.db.util.ws.TextRoomMessage;
import org.kurento.client.Continuation;
import org.kurento.client.IceCandidate;
-import org.kurento.client.MediaFlowState;
import org.kurento.client.MediaProfileSpecType;
import org.kurento.client.MediaType;
import org.kurento.client.RecorderEndpoint;
@@ -62,6 +66,7 @@ public class KStream extends AbstractStream {
private RecorderEndpoint recorder;
private WebRtcEndpoint outgoingMedia = null;
private final ConcurrentMap<String, WebRtcEndpoint> listeners = new ConcurrentHashMap<>();
+ private Optional<CompletableFuture<Object>> flowoutFuture = Optional.empty();
private Long chunkId;
private Type type;
@@ -111,14 +116,28 @@ public class KStream extends AbstractStream {
break;
}
outgoingMedia = createEndpoint(processor, sd.getSid(), sd.getUid());
- outgoingMedia.addMediaSessionTerminatedListener(evt -> log.warn("Media stream terminated"));
+ outgoingMedia.addMediaSessionTerminatedListener(evt -> log.warn("Media stream terminated {}", sd));
outgoingMedia.addMediaFlowOutStateChangeListener(evt -> {
- if (MediaFlowState.NOT_FLOWING == evt.getState()) {
- log.warn("Media FlowOut :: {}", evt.getState());
- if (StreamType.SCREEN == streamType) {
- processor.doStopSharing(sid, uid);
- }
- stopBroadcast(processor);
+ log.info("Media Flow STATE :: {}", evt.getState());
+ switch (evt.getState()) {
+ case NOT_FLOWING:
+ log.warn("FlowOut Future is created");
+ flowoutFuture = Optional.of(new CompletableFuture<>().completeAsync(() -> {
+ log.warn("KStream will be dropped {}", sd);
+ if (StreamType.SCREEN == streamType) {
+ processor.doStopSharing(sid, uid);
+ }
+ stopBroadcast(processor);
+ return null;
+ }, delayedExecutor(getFlowoutTimeout(), TimeUnit.SECONDS)));
+ break;
+ case FLOWING:
+ flowoutFuture.ifPresent(f -> {
+ log.warn("FlowOut Future is canceled");
+ f.cancel(true);
+ flowoutFuture = Optional.empty();
+ });
+ break;
}
});
outgoingMedia.addMediaFlowInStateChangeListener(evt -> log.warn("Media FlowIn :: {}", evt));
diff --git a/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/KurentoHandler.java b/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/KurentoHandler.java
index 475f8d6..2f8a1d8 100644
--- a/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/KurentoHandler.java
+++ b/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/KurentoHandler.java
@@ -78,8 +78,9 @@ public class KurentoHandler {
public static final String TAG_MODE = "mode";
public static final String TAG_ROOM = "roomId";
private static final String HMAC_SHA1_ALGORITHM = "HmacSHA1";
- private final ScheduledExecutorService recheckScheduler = Executors.newScheduledThreadPool(1);
+ private final ScheduledExecutorService kmsRecheckScheduler = Executors.newScheduledThreadPool(1);
public static final String KURENTO_TYPE = "kurento";
+ private static int FLOWOUT_TIMEOUT_SEC = 5;
private long checkTimeout = 120000; //ms
private long objCheckTimeout = 200; //ms
private int watchThreadCount = 10;
@@ -122,7 +123,7 @@ public class KurentoHandler {
client.getServerManager().addObjectCreatedListener(new KWatchDog());
} catch (Exception e) {
log.warn("Fail to create Kurento client, will re-try in {} ms", checkTimeout);
- recheckScheduler.schedule(check, checkTimeout, MILLISECONDS);
+ kmsRecheckScheduler.schedule(check, checkTimeout, MILLISECONDS);
}
};
check.run();
@@ -336,6 +337,14 @@ public class KurentoHandler {
this.turnTtl = turnTtl;
}
+ public void setFlowoutTimeout(int timeout) {
+ FLOWOUT_TIMEOUT_SEC = timeout;
+ }
+
+ static int getFlowoutTimeout() {
+ return FLOWOUT_TIMEOUT_SEC;
+ }
+
private class KConnectionListener implements KurentoConnectionListener {
final String lkuid;
@@ -359,7 +368,7 @@ public class KurentoHandler {
connected = false;
notifyRooms();
destroy();
- recheckScheduler.schedule(check, checkTimeout, MILLISECONDS);
+ kmsRecheckScheduler.schedule(check, checkTimeout, MILLISECONDS);
}
}
diff --git a/openmeetings-web/src/main/webapp/WEB-INF/classes/applicationContext.xml b/openmeetings-web/src/main/webapp/WEB-INF/classes/applicationContext.xml
index b08e56e..39fcbfb 100644
--- a/openmeetings-web/src/main/webapp/WEB-INF/classes/applicationContext.xml
+++ b/openmeetings-web/src/main/webapp/WEB-INF/classes/applicationContext.xml
@@ -154,5 +154,6 @@
p:turnMode="rest"
p:turnTtl="60"
p:objCheckTimeout="200"
+ p:flowoutTimeout="5"
/>
</beans>