Merge pull request #7003 from ant-media/SrtRestream

SRT Restream Endpoint is accepted.

Author: burak-58

src/main/java/io/antmedia/datastore/db/InMemoryDataStore.java

@@ -145,7 +145,7 @@ public boolean removeEndpoint(String id, Endpoint endpoint, boolean checkRTMPUrl
 				for (Iterator<Endpoint> iterator = endPointList.iterator(); iterator.hasNext();) {
 					Endpoint endpointItem = iterator.next();
 					if(checkRTMPUrl) {
-						if (endpointItem.getRtmpUrl().equals(endpoint.getRtmpUrl())) {
+						if (endpointItem.getEndpointUrl().equals(endpoint.getEndpointUrl())) {
 							iterator.remove();
 							result = true;
 							break;

src/main/java/io/antmedia/datastore/db/MapBasedDataStore.java

@@ -197,7 +197,7 @@ public boolean removeEndpoint(String id, Endpoint endpoint, boolean checkRTMPUrl
 						for (Iterator<Endpoint> iterator = endPointList.iterator(); iterator.hasNext();) {
 							Endpoint endpointItem = iterator.next();
 							if (checkRTMPUrl) {
-								if (endpointItem.getRtmpUrl().equals(endpoint.getRtmpUrl())) {
+								if (endpointItem.getEndpointUrl().equals(endpoint.getEndpointUrl())) {
 									iterator.remove();
 									result = true;
 									break;

src/main/java/io/antmedia/datastore/db/MongoStore.java

@@ -513,7 +513,7 @@ public boolean removeEndpoint(String id, Endpoint endpoint, boolean checkRTMPUrl
 						for (Iterator<Endpoint> iterator = endPointList.iterator(); iterator.hasNext();) {
 							Endpoint endpointItem = iterator.next();
 							if(checkRTMPUrl) {
-								if (endpointItem.getRtmpUrl().equals(endpoint.getRtmpUrl())) {
+								if (endpointItem.getEndpointUrl().equals(endpoint.getEndpointUrl())) {
 									iterator.remove();
 									result = true;
 									break;

src/main/java/io/antmedia/datastore/db/types/Endpoint.java

@@ -23,9 +23,19 @@ public class Endpoint {
     @Schema(description = "The service name like facebook, periscope, youtube or generic")
     private String type;
 
+    /**
+     * RTMP Or SRT URL of the endpoint
+     */
+    @Schema(description = "RTMP or SRT URL of the endpoint")
+    private String endpointUrl;
+
     /**
      * RTMP URL of the endpoint
+     *
+     * @deprecated use {@link #endpointUrl}
+     *
      */
+    @Deprecated(since = "3.0" , forRemoval = true)
     @Schema(description = "The RTMP URL of the endpoint")
     private String rtmpUrl;
 
@@ -46,22 +56,28 @@ public Endpoint() {
     public Endpoint(String rtmpUrl, String type, String endpointServiceId, String status) {
         this();
         this.status = status;
-        this.rtmpUrl = rtmpUrl;
+        this.endpointUrl = rtmpUrl;
         this.type = type;
         this.endpointServiceId = endpointServiceId;
     }
 
-    public String getRtmpUrl() {
-        return rtmpUrl;
+    public String getEndpointUrl() {
+        if(endpointUrl==null)
+            return rtmpUrl;
+        return endpointUrl;
     }
 
-    public void setRtmpUrl(String rtmpUrl) {
-        this.rtmpUrl = rtmpUrl;
+    public void setEndpointUrl(String endpointUrl) {
+        this.endpointUrl = endpointUrl;
+        this.rtmpUrl = endpointUrl;
     }
 
     public String getEndpointServiceId() {
         return endpointServiceId;
     }
+    public void setRtmpUrl(String rtmpUrl){
+        this.rtmpUrl = rtmpUrl;
+    }
 
     public void setEndpointServiceId(String endpointServiceId) {
         this.endpointServiceId = endpointServiceId;

…in/java/io/antmedia/muxer/RtmpMuxer.java …ava/io/antmedia/muxer/EndpointMuxer.javasrc/main/java/io/antmedia/muxer/RtmpMuxer.java renamed to src/main/java/io/antmedia/muxer/EndpointMuxer.java src/main/java/io/antmedia/muxer/RtmpMuxer.java renamed to src/main/java/io/antmedia/muxer/EndpointMuxer.java

@@ -19,6 +19,7 @@
 import static org.bytedeco.ffmpeg.global.avutil.AV_ROUND_PASS_MINMAX;
 import static org.bytedeco.ffmpeg.global.avutil.av_rescale_q;
 import static org.bytedeco.ffmpeg.global.avutil.av_rescale_q_rnd;
+import org.bytedeco.ffmpeg.global.avutil;
 
 import java.nio.ByteBuffer;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -34,13 +35,12 @@
 import org.bytedeco.ffmpeg.avformat.AVStream;
 import org.bytedeco.ffmpeg.avutil.AVRational;
 import org.bytedeco.ffmpeg.global.avcodec;
-import org.bytedeco.ffmpeg.global.avutil;
 import org.bytedeco.javacpp.BytePointer;
 import org.bytedeco.javacpp.SizeTPointer;
 
 import io.vertx.core.Vertx;
 
-public class RtmpMuxer extends Muxer {
+public class EndpointMuxer extends Muxer {
 
 	private String url;
 	private volatile boolean trailerWritten = false;
@@ -55,25 +55,40 @@ public class RtmpMuxer extends Muxer {
 	private AtomicBoolean preparedIO = new AtomicBoolean(false);
 	private AtomicBoolean cancelOpenIO = new AtomicBoolean(false);
 
-	public RtmpMuxer(String url, Vertx vertx) {
+	public String muxerType = null;
+
+	public EndpointMuxer(String url, Vertx vertx) {
 		super(vertx);
-		format = "flv";
+		this.format = "flv";
 		this.url = url;
 
-		parseRtmpURL(this.url);
+		parseEndpointURL(this.url);
 	}
-	void parseRtmpURL(String url){
-		if(url == null)
-			return;
-		 // check if app name is present in the URL rtmp://Domain.com/AppName/StreamId
-		String regex = "rtmp(s)?://[a-zA-Z0-9\\.-]+(:[0-9]+)?/([^/]+)/.*";
 
-		Pattern rtmpAppName = Pattern.compile(regex);
-		Matcher checkAppName = rtmpAppName.matcher(url);
+	public String getMuxerType() {
+		return muxerType;
+	}
 
-		if (!checkAppName.matches()) {
-			//this is the fix to send stream for urls without app
-			setOption("rtmp_app","");
+	void parseEndpointURL(String url){
+		if(url == null)
+			return;
+		if(url.startsWith("rtmp")) {
+			format = "flv";
+			muxerType = "rtmp";
+			// check if app name is present in the URL rtmp://Domain.com/AppName/StreamId
+			String regex = "rtmp(s)?://[a-zA-Z0-9\\.-]+(:[0-9]+)?/([^/]+)/.*";
+
+			Pattern rtmpAppName = Pattern.compile(regex);
+			Matcher checkAppName = rtmpAppName.matcher(url);
+
+			if (!checkAppName.matches()) {
+				//this is the fix to send stream for urls without app
+				setOption("rtmp_app", "");
+			}
+		}
+		else if(url.startsWith("srt")){
+			muxerType = "srt";
+			format = "mpegts";
 		}
 	}
 	@Override
@@ -166,7 +181,7 @@ public synchronized boolean prepareIO()
 				{
 					clearResource();
 					setStatus(IAntMediaStreamHandler.BROADCAST_STATUS_FAILED);
-					logger.error("Cannot initializeOutputFormatContextIO for rtmp endpoint:{}", url);
+					logger.error("Cannot initializeOutputFormatContextIO for {} endpoint:{}", muxerType ,url);
 				}
 
 				return null;
@@ -259,7 +274,7 @@ private boolean exitIfCancelled() {
 	public synchronized boolean addVideoStream(int width, int height, AVRational timebase, int codecId, int streamIndex, boolean isAVC, AVCodecParameters codecpar) {
 
 		boolean result = super.addVideoStream(width, height, timebase, codecId, streamIndex, isAVC, codecpar);
-		if (result) 
+		if (result && this.format.equals("flv"))
 		{
 			AVStream outStream = getOutputFormatContext().streams(inputOutputStreamIndexMap.get(streamIndex));
 
@@ -290,7 +305,7 @@ public synchronized void writePacket(AVPacket pkt, final AVRational inputTimebas
 		writeFrameInternal(pkt, inputTimebase, outputTimebase, context, codecType);
 	}
 
-	private synchronized void writeFrameInternal(AVPacket pkt, AVRational inputTimebase, AVRational outputTimebase,
+	public synchronized void writeFrameInternal(AVPacket pkt, AVRational inputTimebase, AVRational outputTimebase,
 			AVFormatContext context, int codecType) 
 	{
 		long pts = pkt.pts();
@@ -321,7 +336,7 @@ private synchronized void writeFrameInternal(AVPacket pkt, AVRational inputTimeb
 					return;
 				}
 
-				while (av_bsf_receive_packet(bsfFilterContextList.get(0), getTmpPacket()) == 0)
+				while ((ret = av_bsf_receive_packet(bsfFilterContextList.get(0), getTmpPacket())) == 0)
 				{
 					if (!headerWritten)
 					{
@@ -410,13 +425,13 @@ public synchronized void writeVideoBuffer(ByteBuffer encodedVideoFrame, long dts
 	{
 
 		if (!isRunning.get() || !registeredStreamIndexList.contains(streamIndex)) {
-			logPacketIssue("Not writing to RTMP muxer because it's not started for {}", url);
+			logPacketIssue("Not writing to {} muxer because it's not started for {}", muxerType,url);
 			return;
 		}
 
 		if (!keyFrameReceived && isKeyFrame) {
 			keyFrameReceived = true;
-			logger.info("Key frame is received to start for rtmp:{}", url);
+			logger.info("Key frame is received to start for {}:{}", muxerType,url);
 		}
 
 		if (keyFrameReceived) {

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

@@ -632,9 +632,9 @@ public static void setUpEndPoints(MuxAdaptor muxAdaptor, Broadcast broadcast, Ve
 			if (endPointList != null && !endPointList.isEmpty()) 
 			{
 				for (Endpoint endpoint : endPointList) {
-					RtmpMuxer rtmpMuxer = new RtmpMuxer(endpoint.getRtmpUrl(), vertx);
-					rtmpMuxer.setStatusListener(muxAdaptor);
-					muxAdaptor.addMuxer(rtmpMuxer);
+					EndpointMuxer endpointMuxer = new EndpointMuxer(endpoint.getEndpointUrl(), vertx);
+					endpointMuxer.setStatusListener(muxAdaptor);
+					muxAdaptor.addMuxer(endpointMuxer);
 				}
 			}
 		}
@@ -2347,7 +2347,7 @@ else if(recordType == RecordType.WEBM) {
 		return muxer;
 	}
 
-	public boolean prepareMuxer(Muxer muxer, int resolutionHeight) 
+	public boolean prepareMuxer(Muxer muxer, int resolutionHeight)
 	{
 		boolean streamAdded = false;
 		muxer.init(scope, streamId, resolutionHeight, getSubfolder(getBroadcast(), getAppSettings()), 0);
@@ -2460,31 +2460,31 @@ public ClientBroadcastStream getBroadcastStream() {
 	}
 
 
-	public Result startRtmpStreaming(String rtmpUrl, int resolutionHeight)
+	public Result startEndpointStreaming(String endpointUrl, int resolutionHeight)
 	{
 		Result result = new Result(false);
-		rtmpUrl = rtmpUrl.replaceAll("[\n\r\t]", "_");
+		endpointUrl = endpointUrl.replaceAll("[\n\r\t]", "_");
 
 		if (!isRecording.get()) 
 		{
-			logger.warn("Start rtmp streaming return false for stream:{} because stream is being prepared", streamId);
-			result.setMessage("Start rtmp streaming return false for stream:"+ streamId +" because stream is being prepared. Try again");			
+			logger.warn("Start endpoint streaming return false for stream:{} because stream is being prepared", streamId);
+			result.setMessage("Start endpoint streaming return false for stream:"+ streamId +" because stream is being prepared. Try again");
 			return result;
 		}
-		logger.info("start rtmp streaming for stream id:{} to {} with requested resolution height{} stream resolution:{}", streamId, rtmpUrl, resolutionHeight, height);
+		logger.info("start endpoint streaming for stream id:{} to {} with requested resolution height{} stream resolution:{}", streamId, endpointUrl, resolutionHeight, height);
 
 		if (resolutionHeight == 0 || resolutionHeight == height) 
 		{
-			RtmpMuxer rtmpMuxer = new RtmpMuxer(rtmpUrl, vertx);
-			rtmpMuxer.setStatusListener(this);
-			if (prepareMuxer(rtmpMuxer, resolutionHeight)) 
+			EndpointMuxer endpointMuxer = new EndpointMuxer(endpointUrl, vertx);
+			endpointMuxer.setStatusListener(this);
+			if (prepareMuxer(endpointMuxer, resolutionHeight))
 			{
 				result.setSuccess(true);
 			}
 			else 
 			{
-				logger.error("RTMP prepare returned false so that rtmp pushing to {} for {} didn't started ", rtmpUrl, streamId);
-				result.setMessage("RTMP prepare returned false so that rtmp pushing to " + rtmpUrl + " for "+ streamId +" didn't started ");
+				logger.error("endpoint prepare returned false so that stream pushing to {} for {} didn't started ", endpointUrl, streamId);
+				result.setMessage("endpoint prepare returned false so that stream pushing to " + endpointUrl + " for "+ streamId +" didn't started ");
 			}
 		}
 
@@ -2555,13 +2555,13 @@ private void tryToRepublish(String url, Long id)
 				logger.info("Health check process failed, trying to republish to the endpoint: {}", url);
 
 				//TODO: 0 as second parameter may cause a problem
-				stopRtmpStreaming(url, 0);
-				startRtmpStreaming(url, height);
+				stopEndpointStreaming(url, 0);
+				startEndpointStreaming(url, height);
 				retryCounter.put(url, tmpRetryCount + 1);
 			}
 			else{
 				logger.info("Exceeded republish retry limit, endpoint {} can't be reached and will be closed" , url);
-				stopRtmpStreaming(url, 0);
+				stopEndpointStreaming(url, 0);
 				sendEndpointErrorNotifyHook(url);
 				retryCounter.remove(url);
 			}
@@ -2623,12 +2623,12 @@ private void updateBroadcastRecord() {
 			for (Iterator iterator = broadcast.getEndPointList().iterator(); iterator.hasNext();) 
 			{
 				Endpoint endpoint = (Endpoint) iterator.next();
-				String statusUpdate = endpointStatusUpdateMap.getOrDefault(endpoint.getRtmpUrl(), null);
+				String statusUpdate = endpointStatusUpdateMap.getOrDefault(endpoint.getEndpointUrl(), null);
 				if (statusUpdate != null) {
 					endpoint.setStatus(statusUpdate);
 				}
 				else {
-					logger.warn("Endpoint is not found to update its status to {} for rtmp url:{}", statusUpdate, endpoint.getRtmpUrl());
+					logger.warn("Endpoint is not found to update its status to {} for rtmp url:{}", statusUpdate, endpoint.getEndpointUrl());
 				}
 			}
 			BroadcastUpdate broadcastUpdate = new BroadcastUpdate();
@@ -2641,38 +2641,38 @@ private void updateBroadcastRecord() {
 		}
 	}
 
-	public RtmpMuxer getRtmpMuxer(String rtmpUrl)
+	public EndpointMuxer getEndpointMuxer(String rtmpUrl)
 	{
-		RtmpMuxer rtmpMuxer = null;
+		EndpointMuxer endpointMuxer = null;
 		synchronized (muxerList)
 		{
 			Iterator<Muxer> iterator = muxerList.iterator();
 			while (iterator.hasNext())
 			{
 				Muxer muxer = iterator.next();
-				if (muxer instanceof RtmpMuxer &&
-						((RtmpMuxer)muxer).getOutputURL().equals(rtmpUrl))
+				if (muxer instanceof EndpointMuxer &&
+						((EndpointMuxer)muxer).getOutputURL().equals(rtmpUrl))
 				{
-					rtmpMuxer = (RtmpMuxer) muxer;
+					endpointMuxer = (EndpointMuxer) muxer;
 					break;
 				}
 			}
 		}
-		return rtmpMuxer;
+		return endpointMuxer;
 	}
 
-	public Result stopRtmpStreaming(String rtmpUrl, int resolutionHeight)
+	public Result stopEndpointStreaming(String endpointUrl, int resolutionHeight)
 	{
 		Result result = new Result(false);
 		if (resolutionHeight == 0 || resolutionHeight == height) 
 		{
-			RtmpMuxer rtmpMuxer = getRtmpMuxer(rtmpUrl);
-			String status = statusMap.getOrDefault(rtmpUrl, null);
-			if (rtmpMuxer != null)
+			EndpointMuxer endpointMuxer = getEndpointMuxer(endpointUrl);
+			String status = statusMap.getOrDefault(endpointUrl, null);
+			if (endpointMuxer != null)
 			{
-				muxerList.remove(rtmpMuxer);
-				statusMap.remove(rtmpUrl);
-				rtmpMuxer.writeTrailer();
+				muxerList.remove(endpointMuxer);
+				statusMap.remove(endpointUrl);
+				endpointMuxer.writeTrailer();
 				result.setSuccess(true);
 			}
 			else if(status == null

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

@@ -187,6 +187,10 @@ public abstract class Muxer {
 		avRationalTimeBase.den(1);
 	}
 
+	public void setFormat(String format) {
+		this.format = format;
+	}
+
 	protected String subFolder = null;
 
 	/**
@@ -235,7 +239,7 @@ public void setStreamIndex(int streamIndex) {
 		public void setKeyFrame(boolean isKeyFrame) {
 			this.keyFrame = isKeyFrame;
 		}
-		
+
 		public ByteBuffer getEncodedVideoFrame() {
 			return encodedVideoFrame;
 		}
@@ -296,6 +300,10 @@ protected Muxer(Vertx vertx) {
 		logger = LoggerFactory.getLogger(this.getClass());
 	}
 
+	public List<AVBSFContext> getBsfFilterContextList() {
+		return bsfFilterContextList;
+	}
+
 	public static File getPreviewFile(IScope scope, String name, String extension) {
 		String appScopeName = ScopeUtils.findApplication(scope).getName();
 		return new File(String.format("%s/webapps/%s/%s", System.getProperty("red5.root"), appScopeName,