Merge remote-tracking branch 'origin/master' into

hlsSegmentFileNameFormat

.github/workflows/test.yml

@@ -73,7 +73,7 @@ jobs:
         run: |
           export RELEASE_VERSION="$(mvn -q -Dexec.executable="echo" -Dexec.args='${project.version}' --non-recursive exec:exec)"
           echo $RELEASE_VERSION
-          mvn clean org.jacoco:jacoco-maven-plugin:prepare-agent package -Dtest=!*/integration/* -Dorg.bytedeco.javacpp.logger.debug=false org.jacoco:jacoco-maven-plugin:report sonar:sonar -Dmaven.javadoc.skip=true --quiet
+          mvn clean org.jacoco:jacoco-maven-plugin:prepare-agent package -Dtest=!*/integration/* -Dorg.bytedeco.javacpp.logger.debug=false org.jacoco:jacoco-maven-plugin:report sonar:sonar -Dmaven.javadoc.skip=true
       
       - name: Show MongoDB Log, Crash Log and Servis Status on failure
         if: failure()

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

@@ -29,6 +29,8 @@
 import java.util.Set;
 import java.util.concurrent.*;
 
+import javax.annotation.Nonnull;
+
 import io.antmedia.filter.JWTFilter;
 import io.antmedia.filter.TokenFilterManager;
 import io.antmedia.statistic.*;
@@ -106,10 +108,10 @@
 import io.antmedia.webrtc.api.IWebRTCClient;
 import io.antmedia.websocket.WebSocketConstants;
 import io.vertx.core.Vertx;
+import io.vertx.core.impl.ConcurrentHashSet;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.dropwizard.MetricsService;
 import jakarta.validation.constraints.NotNull;
-import org.springframework.web.context.WebApplicationContext;
 
 public class AntMediaApplicationAdapter  extends MultiThreadedApplicationAdapter implements IAntMediaStreamHandler, IShutdownListener {
 
@@ -227,6 +229,8 @@ public class AntMediaApplicationAdapter  extends MultiThreadedApplicationAdapter
 	private Random random = new Random();
 
 	private IStatsCollector statsCollector;
+
+	private Set<IAppSettingsUpdateListener> settingsUpdateListenerSet = new ConcurrentHashSet<IAppSettingsUpdateListener>();
 
 	@Override
 	public boolean appStart(IScope app) {
@@ -698,7 +702,9 @@ public void closeBroadcast(String streamId) {
 				}	
 
 				for (IStreamListener listener : streamListeners) {
+					//keep backward compatibility
 					listener.streamFinished(broadcast.getStreamId());
+					listener.streamFinished(broadcast);
 				}
 				logger.info("Leaving closeBroadcast for streamId:{}", streamId);
 			}
@@ -874,6 +880,7 @@ public void startPublish(String streamId, long absoluteStartTimeMs, String publi
 
 				for (IStreamListener listener : streamListeners) {
 					listener.streamStarted(broadcast.getStreamId());
+					listener.streamStarted(broadcast);
 				}
 
 				long videoHeight = 0;
@@ -908,7 +915,7 @@ public void startPublish(String streamId, long absoluteStartTimeMs, String publi
 			}
 			return null;
 
-		});
+		}, false);
 
 
 		if (absoluteStartTimeMs == 0)
@@ -1014,47 +1021,65 @@ public static Broadcast saveUndefinedBroadcast(String streamId, String streamNam
 	}
 
 	@Override
-	public void muxingFinished(final String streamId, File file, long startTime, long duration, int resolution, String previewFilePath, String vodId) {
-		String vodName = file.getName();
-		String filePath = file.getPath();
-		long fileSize = file.length();
-		long systemTime = System.currentTimeMillis();
+	@Deprecated
+	public void muxingFinished(String streamId, File File, long startTime, long duration, int resolution,
+			String previewFilePath, String vodId) 
+	{
+		muxingFinished(getDataStore().get(streamId), streamId, File, startTime, duration, resolution, previewFilePath, vodId);
+	}
 
-		String relativePath = getRelativePath(filePath);
+	@Override
+	public void muxingFinished(Broadcast broadcast, String streamId, File file, long startTime, long duration, int resolution, String previewFilePath, String vodId) {
+
 		String listenerHookURL = null;
 		String streamName = file.getName();
+		String description = null;
+		String metadata = null;
+		String longitude = null;
+		String latitude = null;
+		String altitude = null;
 
-		Broadcast broadcast = getDataStore().get(streamId);
-
-		if(broadcast != null){
+		if (broadcast != null) {
 			listenerHookURL = broadcast.getListenerHookURL();
-			if(broadcast.getName() != null){
+			if(StringUtils.isNotBlank(broadcast.getName())){
 				streamName =  resolution != 0 ? broadcast.getName() + " (" + resolution + "p)" : broadcast.getName();
 			}
+			description = broadcast.getDescription();
+			metadata = broadcast.getMetaData();
+			longitude = broadcast.getLongitude();
+			latitude = broadcast.getLatitude();
+			altitude = broadcast.getAltitude();
+		}
+		else {
+			logger.error("Broadcast is null for muxingFinished for stream: {} it's not supposed to happen", streamId);
 		}
+
+		String vodName = file.getName();
+		String filePath = file.getPath();
+		long fileSize = file.length();
+		long systemTime = System.currentTimeMillis();
+
+		String relativePath = getRelativePath(filePath);
+
+		logger.info("muxing finished for stream: {} with file: {} and duration:{}", streamId, file, duration);
 
 		//We need to get the webhook url explicitly because broadcast may be deleted here
-		if (listenerHookURL == null || listenerHookURL.isEmpty()) {
+		if (StringUtils.isBlank(listenerHookURL)) {
 			// if hook URL is not defined for stream specific, then try to get common one from app
 			listenerHookURL = appSettings.getListenerHookURL();
 		}
 
-		String vodIdFinal;
-		if (vodId != null) {
-			vodIdFinal = vodId;
-		}
-		else {
-			vodIdFinal = RandomStringUtils.randomAlphanumeric(24);
+		if (StringUtils.isBlank(vodId)) {
+			vodId = RandomStringUtils.randomAlphanumeric(24);
 		}
 
-		VoD newVod = new VoD(streamName, streamId, relativePath, vodName, systemTime, startTime, duration, fileSize, VoD.STREAM_VOD, vodIdFinal, previewFilePath);
-		if(broadcast != null){
-			newVod.setDescription(broadcast.getDescription());
-			newVod.setMetadata(broadcast.getMetaData());
-			newVod.setLongitude(broadcast.getLongitude());
-			newVod.setLatitude(broadcast.getLatitude());
-			newVod.setAltitude(broadcast.getAltitude());
-		}
+		VoD newVod = new VoD(streamName, streamId, relativePath, vodName, systemTime, startTime, duration, fileSize, VoD.STREAM_VOD, vodId, previewFilePath);
+		newVod.setDescription(description);
+		newVod.setMetadata(metadata);
+		newVod.setLongitude(longitude);
+		newVod.setLatitude(latitude);
+		newVod.setAltitude(altitude);
+
 
 
 		if (getDataStore().addVod(newVod) == null) {
@@ -1069,9 +1094,8 @@ public void muxingFinished(final String streamId, File file, long startTime, lon
 				|| ((index = vodName.lastIndexOf(".webm")) != -1) )
 		{
 			final String baseName = vodName.substring(0, index);
-			final String metaData = (broadcast != null) ? broadcast.getMetaData() : null;
 			logger.info("Setting timer for calling vod ready hook for stream:{}", streamId);
-			notifyHook(listenerHookURL, streamId, null, HOOK_ACTION_VOD_READY, null, null, baseName, vodIdFinal, metaData, null);
+			notifyHook(listenerHookURL, streamId, null, HOOK_ACTION_VOD_READY, null, null, baseName, vodId, metadata, null);
 		}
 
 		String muxerFinishScript = appSettings.getMuxerFinishScript();
@@ -1307,7 +1331,7 @@ public void trySendClusterPostWithDelay(String url, String clusterCommunicationT
 				}
 				return null;
 
-			});
+			}, false);
 
 
 		});
@@ -2109,9 +2133,22 @@ public synchronized boolean updateSettings(AppSettings newSettings, boolean noti
 		else {
 			logger.warn("Settings cannot be saved for {}", getScope().getName());
 		}
+
+		notifySettingsUpdateListeners(appSettings);
 
 		return result;
 	}
+
+	public void notifySettingsUpdateListeners(AppSettings appSettings) {
+		for (IAppSettingsUpdateListener listener : settingsUpdateListenerSet) {
+			listener.settingsUpdated(appSettings);
+		}
+	}
+
+	@Override
+	public void addSettingsUpdateListener(IAppSettingsUpdateListener listener) {
+		settingsUpdateListenerSet.add(listener);
+	}
 
 	private boolean isEncoderSettingsValid(List<EncoderSettings> encoderSettingsList) {
 		if (encoderSettingsList != null) {
@@ -2251,6 +2288,7 @@ public void setStorageclientSettings(AppSettings settings) {
 		storageClient.setPermission(settings.getS3Permission());
 		storageClient.setStorageClass(settings.getS3StorageClass());
 		storageClient.setCacheControl(settings.getS3CacheControl());
+		storageClient.setTransferBufferSize(settings.getS3TransferBufferSizeInBytes());
 		storageClient.reset();
 	}
 
@@ -2447,7 +2485,7 @@ public void stopPublish(String streamId) {
 		vertx.executeBlocking(() -> {
 			closeBroadcast(streamId);
 			return null;
-		});
+		}, false);
 	}
 
 	public boolean isClusterMode() {

src/main/java/io/antmedia/AppSettings.java

@@ -1806,6 +1806,7 @@ public class AppSettings implements Serializable{
 	public static final String APPLICATION_STATUS_INSTALLATION_FAILED = "installationFailed";
 
 
+
 	/**
 	 * Describes the application installation status. Possible values:
 	 *
@@ -2013,6 +2014,17 @@ public class AppSettings implements Serializable{
 	 */
 	@Value("${s3Permission:${"+SETTINGS_S3_PERMISSION+":public-read}}")
 	private String s3Permission = "public-read";
+
+
+	/**
+	 * S3 Transfer Buffer Size
+	 * This describes to buffer size to keep transferring data. It should be
+	 * bigger than ts segment file size for HLS continuous upload.
+	 * Otherwise chunk update may cannot be retried in case of any network break.
+	 */
+	@Value("${s3TransferBufferSizeInBytes:10000000}")
+	private int s3TransferBufferSizeInBytes = 10000000;
+
 
 	/**
 	 *  HLS Encryption key info file full path.
@@ -4159,4 +4171,12 @@ public long getAppInstallationTime() {
 	public void setAppInstallationTime(long appInstallationTime) {
 		this.appInstallationTime = appInstallationTime;
 	}
+
+	public int getS3TransferBufferSizeInBytes() {
+		return s3TransferBufferSizeInBytes;
+	}
+
+	public void setS3TransferBufferSizeInBytes(int s3TransferBufferSizeInBytes) {
+		this.s3TransferBufferSizeInBytes = s3TransferBufferSizeInBytes;
+	}
 }

src/main/java/io/antmedia/console/AdminApplication.java

@@ -372,7 +372,7 @@ public boolean createApplication(String appName, String warFileFullPath) {
 				currentApplicationCreationProcesses.remove(appName);
 			}
 			return null;
-		});
+		}, false);
 
 		return success;
 

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

@@ -6,6 +6,7 @@
 import org.red5.server.api.scope.IScope;
 
 import io.antmedia.AppSettings;
+import io.antmedia.IAppSettingsUpdateListener;
 import io.antmedia.datastore.db.DataStore;
 import io.antmedia.datastore.db.types.Broadcast;
 import io.antmedia.plugin.api.IFrameListener;
@@ -44,10 +45,28 @@ public interface IAntMediaStreamHandler {
 	 * @param file video file that muxed is finished
 	 * @param duration of the video in milliseconds
 	 * @param resolution height of the video 
+	 * 
+	 * @Deprecated use {@link #muxingFinished(Broadcast, File, long, long, int, String, String)} because Broadcast object may be deleted when this method is called
 	 */
+	@Deprecated
 	public void muxingFinished(String id, File file, long startTime, long duration , int resolution, String path, String vodId);
 
 
+	/**
+	 * Called by some muxer like MP4Muxer
+	 * 
+	 * id actually is the name of the file however in some cases file name and the id may be different
+	 * in some cases like there is already a file with that name
+	 * 
+	 * @param broadcast object that muxed is finished
+	 * @param streamId is the id of the stream
+	 * @param file video file that muxed is finished
+	 * @param duration of the video in milliseconds
+	 * @param resolution height of the video 
+	 * 
+	 */
+	public void muxingFinished(Broadcast broadcast, String streamId, File file, long startTime, long duration , int resolution, String path, String vodId);
+
 	/**
 	 * Update stream quality, speed and number of pending packet size and update time
 	 * in datastore
@@ -229,4 +248,10 @@ public interface IAntMediaStreamHandler {
 	 */
 	public void notifyWebhookForStreamStatus(Broadcast broadcast, int width, int height, long totalByteReceived,
 			int inputQueueSize, double speed);
+
+	/**
+	 * Add listener that is notified when the settings are updated
+	 * @param listener
+	 */
+	public void addSettingsUpdateListener(IAppSettingsUpdateListener listener);
 }

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

@@ -29,6 +29,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.antmedia.FFmpegUtilities;
+import io.antmedia.datastore.db.DataStore;
 import io.antmedia.rest.RestServiceBase;
 
 import org.apache.commons.lang3.exception.ExceptionUtils;
@@ -58,6 +59,7 @@
 import org.springframework.context.ApplicationContext;
 import org.springframework.core.io.Resource;
 
+import io.antmedia.AntMediaApplicationAdapter;
 import io.antmedia.AppSettings;
 import io.vertx.core.Vertx;
 import io.vertx.core.impl.ConcurrentHashSet;
@@ -279,6 +281,8 @@ public long getOriginalFrameTimeMs() {
 
 	private int videoCodecId;
 
+	private IAntMediaStreamHandler appInstance;
+
 
 	protected Muxer(Vertx vertx) {
 		this.vertx = vertx;
@@ -703,6 +707,15 @@ public AppSettings getAppSettings() {
 		ApplicationContext appCtx = context.getApplicationContext();
 		return (AppSettings) appCtx.getBean(AppSettings.BEAN_NAME);
 	}
+
+
+	public AntMediaApplicationAdapter getAppAdaptor() {
+		IContext context = scope.getContext();
+		ApplicationContext appCtx = context.getApplicationContext();
+		AntMediaApplicationAdapter adaptor = (AntMediaApplicationAdapter) appCtx.getBean(AntMediaApplicationAdapter.BEAN_NAME);
+		return adaptor;
+	}
+
 
 	public String getExtendedName(String name, int resolution, int bitrate, String fileNameFormat) {
 		StringBuilder result = new StringBuilder(name);