openmeetings-mediaserver/src/main/java/org/apache/openmeetings/mediaserver/remote/KTestStream.java - openmeetings - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License") +  you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
 package org.apache.openmeetings.mediaserver.remote;

 import static java.util.UUID.randomUUID;
 import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.MODE_TEST;
 import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.PARAM_CANDIDATE;
 import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.TAG_MODE;
 import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.TAG_ROOM;
 import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.sendError;
 import static org.apache.openmeetings.mediaserver.remote.TestStreamProcessor.newTestKurentoMsg;
 import static org.apache.openmeetings.util.OmFileHelper.EXTENSION_WEBM;
 import static org.apache.openmeetings.util.OmFileHelper.TEST_SETUP_PREFIX;
 import static org.apache.openmeetings.util.OmFileHelper.getStreamsDir;

 import java.io.File;
 import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;

 import javax.inject.Inject;

 import org.apache.openmeetings.core.util.WebSocketHelper;
 import org.apache.openmeetings.db.entity.basic.IWsClient;
 import org.apache.openmeetings.util.OmFileHelper;
 import org.apache.wicket.injection.Injector;
 import org.kurento.client.Continuation;
 import org.kurento.client.IceCandidate;
 import org.kurento.client.MediaPipeline;
 import org.kurento.client.MediaProfileSpecType;
 import org.kurento.client.MediaType;
 import org.kurento.client.PlayerEndpoint;
 import org.kurento.client.RecorderEndpoint;
 import org.kurento.client.WebRtcEndpoint;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.github.openjson.JSONObject;

 public class KTestStream extends AbstractStream {
 	private static final Logger log = LoggerFactory.getLogger(KTestStream.class);
 	private static final Map<String, String> TAGS = Map.of(TAG_MODE, MODE_TEST, TAG_ROOM, MODE_TEST);

 	@Inject
 	private KurentoHandler kHandler;
 	@Inject
 	private TestStreamProcessor processor;

 	private MediaPipeline pipeline;
 	private WebRtcEndpoint webRtcEndpoint;
 	private PlayerEndpoint player;
 	private RecorderEndpoint recorder;
 	private String recPath = null;
 	private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
 	private ScheduledFuture<?> recHandle;
 	private int recTime;

 	public KTestStream(IWsClient c, JSONObject msg) {
 		super(null, c.getUid());
 		Injector.get().inject(this);
 		createPipeline(() -> startTestRecording(c, msg));
 	}

 	private void startTestRecording(IWsClient c, JSONObject msg) {
 		webRtcEndpoint = createWebRtcEndpoint(pipeline, null, kHandler.getCertificateType());
 		webRtcEndpoint.connect(webRtcEndpoint);

 		MediaProfileSpecType profile = getProfile(msg);
 		initRecPath();
 		recorder = createRecorderEndpoint(pipeline, recPath, profile);

 		recorder.addRecordingListener(evt -> {
 				recTime = 0;
 				recHandle = scheduler.scheduleAtFixedRate(
 						() -> WebSocketHelper.sendClient(c, newTestKurentoMsg().put("id", "recording").put("time", recTime++))
 						, 0, 1, TimeUnit.SECONDS);
 				scheduler.schedule(() -> {
 						recorder.stop();
 						recHandle.cancel(true);
 					}, 5, TimeUnit.SECONDS);
 			});
 		recorder.addStoppedListener(evt -> {
 				WebSocketHelper.sendClient(c, newTestKurentoMsg().put("id", "recStopped"));
 				releaseRecorder();
 			});
 		switch (profile) {
 			case WEBM:
 				webRtcEndpoint.connect(recorder, MediaType.AUDIO);
 				webRtcEndpoint.connect(recorder, MediaType.VIDEO);
 				break;
 			case WEBM_AUDIO_ONLY:
 				webRtcEndpoint.connect(recorder, MediaType.AUDIO);
 				break;
 			case WEBM_VIDEO_ONLY:
 				webRtcEndpoint.connect(recorder, MediaType.VIDEO);
 				break;
 			default:
 				//no-op
 				break;
 		}

 		String sdpOffer = msg.getString("sdpOffer");
 		String sdpAnswer = webRtcEndpoint.processOffer(sdpOffer);

 		addIceListener(c);

 		WebSocketHelper.sendClient(c, newTestKurentoMsg()
 				.put("id", "startResponse")
 				.put("sdpAnswer", sdpAnswer));
 		webRtcEndpoint.gatherCandidates();
 		recorder.record(new Continuation<Void>() {
 			@Override
 			public void onSuccess(Void result) throws Exception {
 				log.info("Recording started successfully");
 			}

 			@Override
 			public void onError(Throwable cause) throws Exception {
 				sendError(c, "Failed to start recording");
 				log.error("Failed to start recording", cause);
 			}
 		});
 	}

 	public void play(final IWsClient inClient, JSONObject msg) {
 		createPipeline(() -> {
 			webRtcEndpoint = createWebRtcEndpoint(pipeline, true, kHandler.getCertificateType());
 			player = createPlayerEndpoint(pipeline, recPath);
 			player.connect(webRtcEndpoint);
 			webRtcEndpoint.addMediaSessionStartedListener(evt -> {
 					log.info("Media session started {}", evt);
 					player.addErrorListener(event -> {
 							log.info("ErrorEvent for player with uid '{}': {}", inClient.getUid(), event.getDescription());
 							sendPlayEnd(inClient);
 						});
 					player.addEndOfStreamListener(event -> {
 							log.info("EndOfStreamEvent for player with uid '{}'", inClient.getUid());
 							sendPlayEnd(inClient);
 						});
 					player.play();
 				});
 			addIceListener(inClient);

 			WebSocketHelper.sendClient(inClient, newTestKurentoMsg()
 					.put("id", "playResponse")
 					.put("sdpAnswer", webRtcEndpoint.processOffer(msg.getString("sdpOffer"))));

 			webRtcEndpoint.gatherCandidates();
 		});
 	}

 	public void addCandidate(IceCandidate cand) {
 		if (webRtcEndpoint != null) {
 			webRtcEndpoint.addIceCandidate(cand);
 		}
 	}

 	private void createPipeline(Runnable action) {
 		release(false);
 		this.pipeline = kHandler.createPipiline(TAGS, new Continuation<Void>() {
 			@Override
 			public void onSuccess(Void result) throws Exception {
 				action.run();
 			}

 			@Override
 			public void onError(Throwable cause) throws Exception {
 				log.warn("Unable to create pipeline for test stream", cause);
 			}
 		});
 	}

 	private void addIceListener(IWsClient inClient) {
 		webRtcEndpoint.addIceCandidateFoundListener(evt -> {
 				IceCandidate cand = evt.getCandidate();
 				WebSocketHelper.sendClient(inClient, newTestKurentoMsg()
 						.put("id", "iceCandidate")
 						.put(PARAM_CANDIDATE, new JSONObject()
 								.put(PARAM_CANDIDATE, cand.getCandidate())
 								.put("sdpMid", cand.getSdpMid())
 								.put("sdpMLineIndex", cand.getSdpMLineIndex())));
 			});
 	}

 	private void sendPlayEnd(IWsClient inClient) {
 		WebSocketHelper.sendClient(inClient, newTestKurentoMsg().put("id", "playStopped"));
 		releasePlayer();
 	}

 	private static MediaProfileSpecType getProfile(JSONObject msg) {
 		boolean a  = msg.getBoolean("audio")
 				, v = msg.getBoolean("video");
 		if (a && v) {
 			return MediaProfileSpecType.WEBM;
 		} else if (v) {
 			return MediaProfileSpecType.WEBM_VIDEO_ONLY;
 		} else {
 			return MediaProfileSpecType.WEBM_AUDIO_ONLY;
 		}
 	}

 	private void initRecPath() {
 		File f = new File(getStreamsDir(), String.format("%s%s.%s", TEST_SETUP_PREFIX, randomUUID(), EXTENSION_WEBM));
 		recPath = OmFileHelper.getRecUri(f);
 	}

 	private void releaseEndpoint() {
 		if (webRtcEndpoint != null) {
 			webRtcEndpoint.release();
 			webRtcEndpoint = null;
 		}
 	}

 	private void releasePipeline() {
 		if (pipeline != null) {
 			pipeline.release();
 			pipeline = null;
 		}
 	}

 	private void releaseRecorder() {
 		releaseEndpoint();
 		if (recorder != null) {
 			recorder.release();
 			recorder = null;
 		}
 		releasePipeline();
 	}

 	private void releasePlayer() {
 		releaseEndpoint();
 		if (player != null) {
 			player.release();
 			player = null;
 		}
 		releasePipeline();
 	}

 	@Override
 	public void release(boolean remove) {
 		releasePlayer();
 		releaseRecorder();
 		if (remove) {
 			processor.release(this, true);
 		}
 	}
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License") + you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	package org.apache.openmeetings.mediaserver.remote;

	import static java.util.UUID.randomUUID;
	import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.MODE_TEST;
	import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.PARAM_CANDIDATE;
	import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.TAG_MODE;
	import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.TAG_ROOM;
	import static org.apache.openmeetings.mediaserver.remote.KurentoHandler.sendError;
	import static org.apache.openmeetings.mediaserver.remote.TestStreamProcessor.newTestKurentoMsg;
	import static org.apache.openmeetings.util.OmFileHelper.EXTENSION_WEBM;
	import static org.apache.openmeetings.util.OmFileHelper.TEST_SETUP_PREFIX;
	import static org.apache.openmeetings.util.OmFileHelper.getStreamsDir;

	import java.io.File;
	import java.util.Map;
	import java.util.concurrent.Executors;
	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.ScheduledFuture;
	import java.util.concurrent.TimeUnit;

	import javax.inject.Inject;

	import org.apache.openmeetings.core.util.WebSocketHelper;
	import org.apache.openmeetings.db.entity.basic.IWsClient;
	import org.apache.openmeetings.util.OmFileHelper;
	import org.apache.wicket.injection.Injector;
	import org.kurento.client.Continuation;
	import org.kurento.client.IceCandidate;
	import org.kurento.client.MediaPipeline;
	import org.kurento.client.MediaProfileSpecType;
	import org.kurento.client.MediaType;
	import org.kurento.client.PlayerEndpoint;
	import org.kurento.client.RecorderEndpoint;
	import org.kurento.client.WebRtcEndpoint;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.github.openjson.JSONObject;

	public class KTestStream extends AbstractStream {
	private static final Logger log = LoggerFactory.getLogger(KTestStream.class);
	private static final Map<String, String> TAGS = Map.of(TAG_MODE, MODE_TEST, TAG_ROOM, MODE_TEST);

	@Inject
	private KurentoHandler kHandler;
	@Inject
	private TestStreamProcessor processor;

	private MediaPipeline pipeline;
	private WebRtcEndpoint webRtcEndpoint;
	private PlayerEndpoint player;
	private RecorderEndpoint recorder;
	private String recPath = null;
	private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
	private ScheduledFuture<?> recHandle;
	private int recTime;

	public KTestStream(IWsClient c, JSONObject msg) {
	super(null, c.getUid());
	Injector.get().inject(this);
	createPipeline(() -> startTestRecording(c, msg));
	}

	private void startTestRecording(IWsClient c, JSONObject msg) {
	webRtcEndpoint = createWebRtcEndpoint(pipeline, null, kHandler.getCertificateType());
	webRtcEndpoint.connect(webRtcEndpoint);

	MediaProfileSpecType profile = getProfile(msg);
	initRecPath();
	recorder = createRecorderEndpoint(pipeline, recPath, profile);

	recorder.addRecordingListener(evt -> {
	recTime = 0;
	recHandle = scheduler.scheduleAtFixedRate(
	() -> WebSocketHelper.sendClient(c, newTestKurentoMsg().put("id", "recording").put("time", recTime++))
	, 0, 1, TimeUnit.SECONDS);
	scheduler.schedule(() -> {
	recorder.stop();
	recHandle.cancel(true);
	}, 5, TimeUnit.SECONDS);
	});
	recorder.addStoppedListener(evt -> {
	WebSocketHelper.sendClient(c, newTestKurentoMsg().put("id", "recStopped"));
	releaseRecorder();
	});
	switch (profile) {
	case WEBM:
	webRtcEndpoint.connect(recorder, MediaType.AUDIO);
	webRtcEndpoint.connect(recorder, MediaType.VIDEO);
	break;
	case WEBM_AUDIO_ONLY:
	webRtcEndpoint.connect(recorder, MediaType.AUDIO);
	break;
	case WEBM_VIDEO_ONLY:
	webRtcEndpoint.connect(recorder, MediaType.VIDEO);
	break;
	default:
	//no-op
	break;
	}

	String sdpOffer = msg.getString("sdpOffer");
	String sdpAnswer = webRtcEndpoint.processOffer(sdpOffer);

	addIceListener(c);

	WebSocketHelper.sendClient(c, newTestKurentoMsg()
	.put("id", "startResponse")
	.put("sdpAnswer", sdpAnswer));
	webRtcEndpoint.gatherCandidates();
	recorder.record(new Continuation<Void>() {
	@Override
	public void onSuccess(Void result) throws Exception {
	log.info("Recording started successfully");
	}

	@Override
	public void onError(Throwable cause) throws Exception {
	sendError(c, "Failed to start recording");
	log.error("Failed to start recording", cause);
	}
	});
	}

	public void play(final IWsClient inClient, JSONObject msg) {
	createPipeline(() -> {
	webRtcEndpoint = createWebRtcEndpoint(pipeline, true, kHandler.getCertificateType());
	player = createPlayerEndpoint(pipeline, recPath);
	player.connect(webRtcEndpoint);
	webRtcEndpoint.addMediaSessionStartedListener(evt -> {
	log.info("Media session started {}", evt);
	player.addErrorListener(event -> {
	log.info("ErrorEvent for player with uid '{}': {}", inClient.getUid(), event.getDescription());
	sendPlayEnd(inClient);
	});
	player.addEndOfStreamListener(event -> {
	log.info("EndOfStreamEvent for player with uid '{}'", inClient.getUid());
	sendPlayEnd(inClient);
	});
	player.play();
	});
	addIceListener(inClient);

	WebSocketHelper.sendClient(inClient, newTestKurentoMsg()
	.put("id", "playResponse")
	.put("sdpAnswer", webRtcEndpoint.processOffer(msg.getString("sdpOffer"))));

	webRtcEndpoint.gatherCandidates();
	});
	}

	public void addCandidate(IceCandidate cand) {
	if (webRtcEndpoint != null) {
	webRtcEndpoint.addIceCandidate(cand);
	}
	}

	private void createPipeline(Runnable action) {
	release(false);
	this.pipeline = kHandler.createPipiline(TAGS, new Continuation<Void>() {
	@Override
	public void onSuccess(Void result) throws Exception {
	action.run();
	}

	@Override
	public void onError(Throwable cause) throws Exception {
	log.warn("Unable to create pipeline for test stream", cause);
	}
	});
	}

	private void addIceListener(IWsClient inClient) {
	webRtcEndpoint.addIceCandidateFoundListener(evt -> {
	IceCandidate cand = evt.getCandidate();
	WebSocketHelper.sendClient(inClient, newTestKurentoMsg()
	.put("id", "iceCandidate")
	.put(PARAM_CANDIDATE, new JSONObject()
	.put(PARAM_CANDIDATE, cand.getCandidate())
	.put("sdpMid", cand.getSdpMid())
	.put("sdpMLineIndex", cand.getSdpMLineIndex())));
	});
	}

	private void sendPlayEnd(IWsClient inClient) {
	WebSocketHelper.sendClient(inClient, newTestKurentoMsg().put("id", "playStopped"));
	releasePlayer();
	}

	private static MediaProfileSpecType getProfile(JSONObject msg) {
	boolean a = msg.getBoolean("audio")
	, v = msg.getBoolean("video");
	if (a && v) {
	return MediaProfileSpecType.WEBM;
	} else if (v) {
	return MediaProfileSpecType.WEBM_VIDEO_ONLY;
	} else {
	return MediaProfileSpecType.WEBM_AUDIO_ONLY;
	}
	}

	private void initRecPath() {
	File f = new File(getStreamsDir(), String.format("%s%s.%s", TEST_SETUP_PREFIX, randomUUID(), EXTENSION_WEBM));
	recPath = OmFileHelper.getRecUri(f);
	}

	private void releaseEndpoint() {
	if (webRtcEndpoint != null) {
	webRtcEndpoint.release();
	webRtcEndpoint = null;
	}
	}

	private void releasePipeline() {
	if (pipeline != null) {
	pipeline.release();
	pipeline = null;
	}
	}

	private void releaseRecorder() {
	releaseEndpoint();
	if (recorder != null) {
	recorder.release();
	recorder = null;
	}
	releasePipeline();
	}

	private void releasePlayer() {
	releaseEndpoint();
	if (player != null) {
	player.release();
	player = null;
	}
	releasePipeline();
	}

	@Override
	public void release(boolean remove) {
	releasePlayer();
	releaseRecorder();
	if (remove) {
	processor.release(this, true);
	}
	}
	}