[openmeetings] branch master updated: [OPENMEETINGS-2301] code is improved a bit (original issue still reproducible)

[so...@apache.org]

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 8b0c13e  [OPENMEETINGS-2301] code is improved a bit (original issue still reproducible)
8b0c13e is described below

commit 8b0c13e05f0d6da1f561b54ce2fe8b4b2c7192de
Author: Maxim Solodovnik <so...@gmail.com>
AuthorDate: Mon May 4 21:08:23 2020 +0700

    [OPENMEETINGS-2301] code is improved a bit (original issue still reproducible)
---
 .../openmeetings/core/remote/KurentoHandler.java   | 37 +++++++++++-----------
 .../openmeetings/core/remote/StreamProcessor.java  | 12 +++++--
 2 files changed, 28 insertions(+), 21 deletions(-)

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 348f3e3..0276767 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
@@ -122,16 +122,19 @@ public class KurentoHandler {
 	public void init() {
 		check = () -> {
 			try {
-				if (isConnected()) {
+				if (client != null) {
 					return;
 				}
+				log.debug("Reconnecting KMS");
 				kuid = randomUUID().toString();
-				client = KurentoClient.create(kurentoWsUrl, new KConnectionListener(kuid));
+				client = KurentoClient.create(kurentoWsUrl, new KConnectionListener());
 				client.getServerManager().addObjectCreatedListener(new KWatchDogCreate());
 				client.getServerManager().addObjectDestroyedListener(event ->
 					log.debug("Kurento::ObjectDestroyedEvent objectId {}, tags {}, source {}", event.getObjectId(), event.getTags(), event.getSource())
 				);
 			} catch (Exception e) {
+				connected = false;
+				clean();
 				log.warn("Fail to create Kurento client, will re-try in {} ms", checkTimeout);
 			}
 		};
@@ -146,15 +149,19 @@ public class KurentoHandler {
 
 	private void clean() {
 		if (client != null) {
-			kuid = randomUUID().toString(); // will be changed to prevent double events
-			client.destroy();
+			KurentoClient copy = client;
+			client = null;
+			testProcessor.destroy();
+			streamProcessor.destroy();
 			for (Entry<Long, KRoom> e : rooms.entrySet()) {
 				e.getValue().close();
 			}
-			testProcessor.destroy();
-			streamProcessor.destroy();
 			rooms.clear();
-			client = null;
+			if (copy != null && !copy.isClosed()) {
+				log.debug("Client will destroyed ...");
+				copy.destroy();
+				log.debug(".... Client is destroyed");
+			}
 		}
 	}
 
@@ -223,7 +230,7 @@ public class KurentoHandler {
 	}
 
 	public void remove(IWsClient c) {
-		if (!isConnected() ||c == null) {
+		if (!isConnected() || c == null) {
 			return;
 		}
 		if (!(c instanceof Client)) {
@@ -365,12 +372,6 @@ public class KurentoHandler {
 	}
 
 	private class KConnectionListener implements KurentoConnectionListener {
-		final String lkuid;
-
-		private KConnectionListener(final String lkuid) {
-			this.lkuid = lkuid;
-		}
-
 		private void notifyRooms() {
 			WebSocketHelper.sendServer(new TextRoomMessage(null, new User(), RoomMessage.Type.KURENTO_STATUS, new JSONObject().put("connected", isConnected()).toString()));
 		}
@@ -383,11 +384,9 @@ public class KurentoHandler {
 		@Override
 		public void disconnected() {
 			log.info("Kurento disconnected");
-			if (lkuid.equals(kuid)) {
-				connected = false;
-				notifyRooms();
-				clean();
-			}
+			connected = false;
+			notifyRooms();
+			clean();
 		}
 
 		@Override
diff --git a/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/StreamProcessor.java b/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/StreamProcessor.java
index 4fa49fe..729e77a 100644
--- a/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/StreamProcessor.java
+++ b/openmeetings-core/src/main/java/org/apache/openmeetings/core/remote/StreamProcessor.java
@@ -174,8 +174,11 @@ public class StreamProcessor implements IStreamProcessor {
 	}
 
 	private void handleBroadcastStarted(Client c, final String uid, JSONObject msg) {
+		if (!kHandler.isConnected()) {
+			return;
+		}
 		StreamDesc sd = c.getStream(uid);
-		KStream sender= getByUid(uid);
+		KStream sender = getByUid(uid);
 		try {
 			if (sender == null) {
 				KRoom room = kHandler.getRoom(c.getRoomId());
@@ -230,6 +233,9 @@ public class StreamProcessor implements IStreamProcessor {
 
 	public void toggleActivity(Client c, Activity a) {
 		log.info("PARTICIPANT {}: trying to toggle activity {}", c, a);
+		if (!kHandler.isConnected()) {
+			return;
+		}
 
 		if (activityAllowed(c, a, c.getRoom())) {
 			boolean wasBroadcasting = isBroadcasting(c);
@@ -567,6 +573,8 @@ public class StreamProcessor implements IStreamProcessor {
 
 	@Override
 	public void destroy() {
-		streamByUid.clear();
+		for (Map.Entry<String, KStream> e : streamByUid.entrySet()) {
+			release(e.getValue(), true);
+		}
 	}
 }