New webhooks playStart, playStop, recordStart issue #4666

1

@@ -0,0 +1,14 @@
+New webhooks playStart, playStop, recordStart issue #4666
+
+# Please enter the commit message for your changes. Lines starting
+# with '#' will be ignored, and an empty message aborts the commit.
+#
+# On branch newWebHooks
+# Your branch is up to date with 'origin/newWebHooks'.
+#
+# Changes to be committed:
+#	modified:   src/main/java/io/antmedia/AntMediaApplicationAdapter.java
+#	modified:   src/main/java/io/antmedia/statistic/ViewerStats.java
+#	modified:   src/test/java/io/antmedia/test/AntMediaApplicationAdaptorUnitTest.java
+#	modified:   src/test/java/io/antmedia/test/statistic/DashViewerStatsTest.java
+#

src/main/java/io/antmedia/AntMediaApplicationAdapter.java

@@ -15,7 +15,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.regex.Pattern;
 import java.util.Queue;
 import java.util.Set;
 
@@ -47,8 +46,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.google.errorprone.annotations.NoAllocation;
-
 import io.antmedia.cluster.ClusterNode;
 import io.antmedia.cluster.IClusterNotifier;
 import io.antmedia.datastore.db.DataStore;
@@ -65,7 +62,6 @@
 import io.antmedia.plugin.api.IFrameListener;
 import io.antmedia.plugin.api.IPacketListener;
 import io.antmedia.plugin.api.IStreamListener;
-import io.antmedia.plugin.api.StreamParametersInfo;
 import io.antmedia.rest.RestServiceBase;
 import io.antmedia.rest.model.Result;
 import io.antmedia.security.AcceptOnlyStreamsInDataStore;
@@ -99,6 +95,9 @@ public class AntMediaApplicationAdapter  extends MultiThreadedApplicationAdapter
 	public static final String HOOK_ACTION_PUBLISH_TIMEOUT_ERROR = "publishTimeoutError";
 	public static final String HOOK_ACTION_ENCODER_NOT_OPENED_ERROR =  "encoderNotOpenedError";
 	public static final String HOOK_ACTION_ENDPOINT_FAILED = "endpointFailed";
+	public static final String HOOK_ACTION_START_PLAY = "playStart";
+	public static final String HOOK_ACTION_STOP_PLAY = "playStop";
+	public static final String HOOK_ACTION_START_RECORD = "recordStart";
 
 	public static final String STREAMS = "streams";
 
@@ -488,7 +487,7 @@ public void closeBroadcast(String streamId) {
 					final String name = broadcast.getName();
 					final String category = broadcast.getCategory();
 					logger.info("Setting timer to call live stream ended hook for stream:{}",streamId );
-					vertx.runOnContext(e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_END_LIVE_STREAM, name, category, null, null, null));
+					vertx.runOnContext(e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_END_LIVE_STREAM, name, category, null, null, null, null));
 				}
 
 				if (broadcast.isZombi()) {
@@ -539,13 +538,69 @@ public void resetDASHStats(String streamId) {
 		}
 	}
 
+	public void sendStartPlayWebHook(final String streamId, final String viewerId){
+		final Broadcast broadcast = getDataStore().get(streamId);
+		if(broadcast == null){
+			return;
+		}
+		final String listenerHookURL = broadcast.getListenerHookURL();
+		if (listenerHookURL == null || listenerHookURL.isEmpty()) {
+			return;
+		}
+		final String name = broadcast.getName();
+		final String category = broadcast.getCategory();
+		logger.info("Setting timer to call viewer play started hook for stream:{}", streamId);
+		vertx.setTimer(10, e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_START_PLAY, name, category,
+				null, null, viewerId, null));
+	}
+
+	public void sendStopPlayWebHook(final String streamId, final String viewerId){
+		final Broadcast broadcast = getDataStore().get(streamId);
+		if(broadcast == null){
+			return;
+		}
+		final String listenerHookURL = broadcast.getListenerHookURL();
+		if (listenerHookURL == null || listenerHookURL.isEmpty()) {
+			return;
+		}
+		final String name = broadcast.getName();
+		final String category = broadcast.getCategory();
+		logger.info("Setting timer to call viewer play stopped hook for stream:{}", streamId);
+		vertx.setTimer(10, e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_STOP_PLAY, name, category,
+				null, null, viewerId, null));
+	}
+
+	public void sendStartRecordWebHook(final String streamId){
+		final Broadcast broadcast = getDataStore().get(streamId);
+		if(broadcast == null){
+			return;
+		}
+		final String listenerHookURL = broadcast.getListenerHookURL();
+		if (listenerHookURL == null || listenerHookURL.isEmpty()) {
+			return;
+		}
+		final String name = broadcast.getName();
+		final String category = broadcast.getCategory();
+		logger.info("Setting timer to call stream start recording hook for stream:{}", streamId);
+		vertx.setTimer(10, e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_START_RECORD, name, category,
+				null, null, null, null));
+	}
+
+	private String getRtmpViewerId(){
+		return "rtmp_" + RandomStringUtils.randomNumeric(8);
+	}
+
 	@Override
 	public void streamPlayItemPlay(ISubscriberStream stream, IPlayItem item, boolean isLive) {
-		vertx.setTimer(1, l -> getDataStore().updateRtmpViewerCount(item.getName(), true));
+		final String streamId = item.getName();
+		sendStartPlayWebHook(streamId, getRtmpViewerId());
+		vertx.setTimer(1, l -> getDataStore().updateRtmpViewerCount(streamId, true));
 	}
 	@Override
 	public void streamPlayItemStop(ISubscriberStream stream, IPlayItem item) {
-		vertx.setTimer(1, l -> getDataStore().updateRtmpViewerCount(item.getName(), false));
+		final String streamId = item.getName();
+		sendStopPlayWebHook(streamId, getRtmpViewerId());
+		vertx.setTimer(1, l -> getDataStore().updateRtmpViewerCount(streamId, false));
 	}
 
 	@Override
@@ -557,7 +612,6 @@ public void streamSubscriberClose(ISubscriberStream stream) {
 	public void startPublish(String streamId, long absoluteStartTimeMs, String publishType) {
 		vertx.executeBlocking( handler -> {
 			try {
-
 				Broadcast broadcast = updateBroadcastStatus(streamId, absoluteStartTimeMs, publishType, getDataStore().get(streamId));
 
 				final String listenerHookURL = getListenerHookURL(broadcast);
@@ -567,7 +621,13 @@ public void startPublish(String streamId, long absoluteStartTimeMs, String publi
 					final String category = broadcast.getCategory();
 					logger.info("Setting timer to call live stream started hook for stream:{}",streamId );
 					vertx.setTimer(10, e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_START_LIVE_STREAM, name, category,
-							null, null, null));
+							null, null, null, null));
+				}
+
+				if ((broadcast.getMp4Enabled() == MuxAdaptor.RECORDING_ENABLED_FOR_STREAM || broadcast.getWebMEnabled() == MuxAdaptor.RECORDING_ENABLED_FOR_STREAM)
+						|| (appSettings.isMp4MuxingEnabled() || appSettings.isWebMMuxingEnabled())
+				) {
+					sendStartRecordWebHook(streamId);
 				}
 
 				int ingestingStreamLimit = appSettings.getIngestingStreamLimit();
@@ -735,7 +795,7 @@ public void muxingFinished(final String streamId, File file, long startTime, lon
 			final String baseName = vodName.substring(0, index);
 			String finalListenerHookURL = listenerHookURL;
 			logger.info("Setting timer for calling vod ready hook for stream:{}", streamId);
-			vertx.runOnContext(e ->	notifyHook(finalListenerHookURL, streamId, HOOK_ACTION_VOD_READY, null, null, baseName, vodIdFinal, null));
+			vertx.runOnContext(e ->	notifyHook(finalListenerHookURL, streamId, HOOK_ACTION_VOD_READY, null, null, baseName, vodIdFinal, null, null));
 		}
 
 		String muxerFinishScript = appSettings.getMuxerFinishScript();
@@ -806,7 +866,7 @@ public static String getRelativePath(String filePath){
 	 * @return
 	 */
 	public StringBuilder notifyHook(String url, String id, String action, String streamName, String category,
-			String vodName, String vodId, String metadata) {
+			String vodName, String vodId, String viewerId, String metadata) {
 		StringBuilder response = null;
 		logger.info("Running notify hook url:{} stream id: {} action:{} vod name:{} vod id:{}", url, id, action, vodName, vodId);
 		if (url != null && url.length() > 0) {
@@ -829,6 +889,10 @@ public StringBuilder notifyHook(String url, String id, String action, String str
 				variables.put("vodId", vodId);
 			}
 
+			if(viewerId != null){
+				variables.put("viewerId", viewerId);
+			}
+
 			if (metadata != null) {
 				variables.put("metadata", metadata);
 			}
@@ -1304,7 +1368,7 @@ public synchronized void incrementEncoderNotOpenedError(String streamId) {
 				final String name = broadcast.getName();
 				final String category = broadcast.getCategory();
 				logger.info("Setting timer to call encoder not opened error for stream:{}", streamId);
-				vertx.runOnContext(e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_ENCODER_NOT_OPENED_ERROR, name, category, null, null, null));
+				vertx.runOnContext(e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_ENCODER_NOT_OPENED_ERROR, name, category, null, null, null,null));
 			}
 		}
 	}
@@ -1329,7 +1393,7 @@ public synchronized void publishTimeoutError(String streamId) {
 				final String name = broadcast.getName();
 				final String category = broadcast.getCategory();
 				logger.info("Setting timer to call hook that means live stream is not started to the publish timeout for stream:{}", streamId);
-				vertx.runOnContext(e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_PUBLISH_TIMEOUT_ERROR, name, category, null, null, null));
+				vertx.runOnContext(e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_PUBLISH_TIMEOUT_ERROR, name, category, null, null, null, null));
 			}
 		}
 	}
@@ -1441,7 +1505,6 @@ private boolean isEncoderSettingsValid(List<EncoderSettings> encoderSettingsList
 	/**
 	 * 
 	 * @param newSettings
-	 * @param checkUpdateTime
 	 * @return true if timing is valid, false if it is invalid
 	 */
 	public boolean isIncomingTimeValid(AppSettings newSettings) 
@@ -1686,7 +1749,7 @@ public void endpointFailedUpdate(String streamId, String url) {
 				logger.info("Setting timer to call rtmp endpoint failed hook for stream:{}", streamId);
 				JSONObject jsonObject = new JSONObject();
 				jsonObject.put("rtmp-url", url);
-				vertx.runOnContext(e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_ENDPOINT_FAILED, name, category, null, null, jsonObject.toJSONString()));
+				vertx.runOnContext(e -> notifyHook(listenerHookURL, streamId, HOOK_ACTION_ENDPOINT_FAILED, name, category, null, null, null, jsonObject.toJSONString()));
 			}
 		}
 	}

src/main/java/io/antmedia/filter/AbstractFilter.java

@@ -9,6 +9,7 @@
 import javax.servlet.FilterConfig;
 import javax.servlet.ServletException;
 
+import io.antmedia.AntMediaApplicationAdapter;
 import org.apache.catalina.util.NetMask;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -154,5 +155,16 @@ public Broadcast getBroadcast(String streamId) {
 		}
 		return broadcast;
 	}
+
+	protected AntMediaApplicationAdapter getAntMediaApplicationAdapter(){
+		AntMediaApplicationAdapter antMediaApplicationAdapter = null;
+		ApplicationContext context = getAppContext();
+		if (context != null)
+		{
+			antMediaApplicationAdapter= (AntMediaApplicationAdapter)context.getBean(AntMediaApplicationAdapter.BEAN_NAME);
+		}
+		return antMediaApplicationAdapter;
+
+	}
 
 }

src/main/java/io/antmedia/filter/DashStatisticsFilter.java

@@ -24,8 +24,6 @@ public class DashStatisticsFilter extends AbstractFilter {
 	@Override
 	public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
 			throws IOException, ServletException {
-
-
 		HttpServletRequest httpRequest =(HttpServletRequest)request;
 
 		String method = httpRequest.getMethod();
@@ -52,8 +50,7 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha
 				logger.debug("req ip {} session id {} stream id {} status {}", request.getRemoteHost(), sessionId, streamId, status);
 				IStreamStats stats = getStreamStats(DashViewerStats.BEAN_NAME);
 				if (stats != null) {
-					stats.registerNewViewer(streamId, sessionId, subscriberId);
-
+					stats.registerNewViewer(streamId, sessionId, subscriberId, getAntMediaApplicationAdapter());
 				}
 			}
 		}

src/main/java/io/antmedia/filter/HlsStatisticsFilter.java

@@ -52,8 +52,7 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha
 				logger.debug("req ip {} session id {} stream id {} status {}", request.getRemoteHost(), sessionId, streamId, status);
 				IStreamStats stats = getStreamStats(HlsViewerStats.BEAN_NAME);
 				if (stats != null) {
-					stats.registerNewViewer(streamId, sessionId, subscriberId);
-
+					stats.registerNewViewer(streamId, sessionId, subscriberId, getAntMediaApplicationAdapter());
 				}
 			}
 		}

src/main/java/io/antmedia/muxer/MuxAdaptor.java

@@ -1681,7 +1681,6 @@ private Muxer addMp4Muxer() {
 	 * @return
 	 */
 	public RecordMuxer startRecording(RecordType recordType) {
-
 		if (!isRecording.get()) {
 			logger.warn("Starting recording return false for stream:{} because stream is being prepared", streamId);
 			return null;
@@ -1692,7 +1691,6 @@ public RecordMuxer startRecording(RecordType recordType) {
 			return null;
 		}
 
-
 		RecordMuxer muxer = null;
 		if(recordType == RecordType.MP4) {
 			Mp4Muxer mp4Muxer = createMp4Muxer();
@@ -1862,10 +1860,15 @@ public Result startRtmpStreaming(String rtmpUrl, int resolutionHeight)
 	}
 
 	public void sendEndpointErrorNotifyHook(String url){
+		AntMediaApplicationAdapter adaptor = getAntMediaApplicationAdaptor();
+		adaptor.endpointFailedUpdate(this.streamId, url);
+	}
+
+	protected AntMediaApplicationAdapter getAntMediaApplicationAdaptor(){
 		IContext context = MuxAdaptor.this.scope.getContext();
 		ApplicationContext appCtx = context.getApplicationContext();
 		AntMediaApplicationAdapter adaptor = (AntMediaApplicationAdapter) appCtx.getBean(AntMediaApplicationAdapter.BEAN_NAME);
-		adaptor.endpointFailedUpdate(this.streamId, url);
+		return adaptor;
 	}
 
 	/**

src/main/java/io/antmedia/statistic/DashViewerStats.java

@@ -1,6 +1,7 @@
 package io.antmedia.statistic;
 
 
+import io.antmedia.AntMediaApplicationAdapter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationContext;
@@ -28,11 +29,11 @@ public void setApplicationContext(ApplicationContext applicationContext)  {
 
 		AppSettings settings = (AppSettings)applicationContext.getBean(AppSettings.BEAN_NAME);
 		timeoutMS = getTimeoutMSFromSettings(settings, timeoutMS, DASH_TYPE);
-
+		final AntMediaApplicationAdapter antMediaApplicationAdapter = (AntMediaApplicationAdapter)applicationContext.getBean(AntMediaApplicationAdapter.BEAN_NAME);
 		vertx.setPeriodic(DEFAULT_TIME_PERIOD_FOR_VIEWER_COUNT, yt-> 
 		{
 			synchronized (lock) {
-				updateViewerCountProcess(DASH_TYPE);
+				updateViewerCountProcess(DASH_TYPE, antMediaApplicationAdapter);
 			}
 		});	
 	}

src/main/java/io/antmedia/statistic/HlsViewerStats.java

@@ -1,5 +1,6 @@
 package io.antmedia.statistic;
 
+import io.antmedia.AntMediaApplicationAdapter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationContext;
@@ -28,11 +29,12 @@ public void setApplicationContext(ApplicationContext applicationContext)  {
 
 		AppSettings settings = (AppSettings)applicationContext.getBean(AppSettings.BEAN_NAME);
 		timeoutMS = getTimeoutMSFromSettings(settings, timeoutMS, HLS_TYPE);
-
+		final AntMediaApplicationAdapter antMediaApplicationAdapter = (AntMediaApplicationAdapter)applicationContext.getBean(AntMediaApplicationAdapter.BEAN_NAME);
+
 		vertx.setPeriodic(DEFAULT_TIME_PERIOD_FOR_VIEWER_COUNT, yt-> 
 		{
 			synchronized (lock) {
-				updateViewerCountProcess(HLS_TYPE);
+				updateViewerCountProcess(HLS_TYPE, antMediaApplicationAdapter);
 			}
 		});	
 	}

src/main/java/io/antmedia/statistic/IStreamStats.java

@@ -1,13 +1,16 @@
 package io.antmedia.statistic;
 
+import io.antmedia.AntMediaApplicationAdapter;
+
 public interface IStreamStats {
 
 	/**
 	 * Register a new viewer to a stream
+	 *
 	 * @param streamId
 	 * @param sessionId
 	 */
-	void registerNewViewer(String streamId, String sessionId, String subscriberId);
+	void registerNewViewer(String streamId, String sessionId, String subscriberId, AntMediaApplicationAdapter antMediaApplicationAdapter);
 
 
 	/**