Merge pull request #5162 from ant-media/add_hls_upload_servlet

Upload HLS files to S3 on the fly

Author: mekya

pom.xml

@@ -600,6 +600,18 @@
             <version>${mockito-core.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-module-junit4</artifactId>
+            <version>2.0.9</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-api-mockito2</artifactId>
+            <version>2.0.9</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>javax.websocket</groupId>
             <artifactId>javax.websocket-api</artifactId>

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

@@ -1606,6 +1606,7 @@ public void setStorageclientSettings(AppSettings settings) {
 		storageClient.setEnabled(settings.isS3RecordingEnabled());
 		storageClient.setPermission(settings.getS3Permission());
 		storageClient.setStorageClass(settings.getS3StorageClass());
+		storageClient.setCacheControl(settings.getS3CacheControl());
 		storageClient.reset();
 	}
 

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

@@ -308,6 +308,7 @@ public class AppSettings implements Serializable{
 	public static final String SETTINGS_S3_BUCKET_NAME = "settings.s3BucketName";
 	public static final String SETTINGS_S3_ENDPOINT = "settings.s3Endpoint";
 	public static final String SETTINGS_S3_PERMISSION = "settings.s3Permission";
+	public static final String SETTINGS_S3_CACHE_CONTROL = "settings.s3CacheControl";
 	public static final String SETTINGS_ENABLE_TIME_TOKEN_PLAY = "settings.enableTimeTokenForPlay";
 	public static final String SETTINGS_ENABLE_TIME_TOKEN_PUBLISH = "settings.enableTimeTokenForPublish";
 
@@ -1421,6 +1422,12 @@ public class AppSettings implements Serializable{
 	@Value( "${"+SETTINGS_S3_ENDPOINT+":#{null}}" )
 	private String s3Endpoint;
 
+	/**
+	 * S3 Cache Control Metadata
+	 */
+	@Value( "${"+SETTINGS_S3_CACHE_CONTROL+":no-store, no-cache, must-revalidate, max-age=0}" )
+	private String s3CacheControl = "no-store, no-cache, must-revalidate, max-age=0";
+
 	/*
 	 * The permission to use in uploading the files to the S3. 
 	 * Following values are accepted. Default value is public-read
@@ -1607,7 +1614,7 @@ public boolean isWriteStatsToDatastore() {
 	 * 
 	 * In initialization no matter if spring or field definition is effective, the important thing is that having some random value
 	 */
-	@Value( "${"+SETTINGS_CLUSTER_COMMUNICATION_KEY+ ":+ #{ T(org.apache.commons.lang3.RandomStringUtils).randomAlphanumeric(32)}" )
+	@Value( "${"+SETTINGS_CLUSTER_COMMUNICATION_KEY+ ":#{ T(org.apache.commons.lang3.RandomStringUtils).randomAlphanumeric(32)}" )
 	private String clusterCommunicationKey = RandomStringUtils.randomAlphanumeric(32);
 
 	public void setWriteStatsToDatastore(boolean writeStatsToDatastore) {
@@ -2830,6 +2837,14 @@ public void setS3Endpoint(String s3Endpoint) {
 		this.s3Endpoint = s3Endpoint;
 	}
 
+	public String getS3CacheControl() {
+		return s3CacheControl;
+	}
+
+	public void setS3CacheControl(String s3CacheControl) {
+		this.s3CacheControl = s3CacheControl;
+	}
+
 	public void setDashHttpEndpoint(String dashHttpEndpoint) {
 		this.dashHttpEndpoint = dashHttpEndpoint;
 	}

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

@@ -158,9 +158,9 @@ public void init(IScope scope, String name, int resolutionHeight, String subFold
 	@Override
 	public String getOutputURL() 
 	{
-		if (httpEndpoint != null) 
+		if (httpEndpoint != null )
 		{
-			return httpEndpoint + File.separator + streamId  + extension;
+			return httpEndpoint + File.separator + initialResourceNameWithoutExtension  + extension;
 		}
 		return super.getOutputURL();
 	}

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

@@ -252,10 +252,10 @@ public PacketTime(long packetTimeMs, long systemTimeMs) {
 		TIME_BASE_FOR_MS.num(1);
 		TIME_BASE_FOR_MS.den(1000);
 	}
-	
+
 	private AVRational videoTimeBase = TIME_BASE_FOR_MS;
 	private AVRational audioTimeBase = TIME_BASE_FOR_MS;
-	
+
 	//NOSONAR because we need to keep the reference of the field
 	private AVChannelLayout channelLayout;
 
@@ -301,7 +301,7 @@ protected MuxAdaptor(ClientBroadcastStream clientBroadcastStream) {
 
 		this.broadcastStream = clientBroadcastStream;
 	}
-	
+
 	public boolean addMuxer(Muxer muxer) {
 		return addMuxer(muxer, 0);
 	}
@@ -512,7 +512,7 @@ else if (broadcast.getWebMEnabled() == RECORDING_ENABLED_FOR_STREAM)
 		}
 
 	}
-	
+
 	public static void setUpEndPoints(MuxAdaptor muxAdaptor, Broadcast broadcast, Vertx vertx) 
 	{
 		if (broadcast != null) {
@@ -536,17 +536,17 @@ public AVCodecParameters getAudioCodecParameters() {
 		if (audioDataConf != null && audioCodecParameters == null) 
 		{
 			AACConfigParser aacParser = new AACConfigParser(audioDataConf, 0);
-						
+
 			if (!aacParser.isErrorOccured()) 
 			{
 				audioCodecParameters = new AVCodecParameters();
 				audioCodecParameters.sample_rate(aacParser.getSampleRate());
-				
+
 				channelLayout = new AVChannelLayout();
 				av_channel_layout_default(channelLayout, aacParser.getChannelCount());
-				
+
 				audioCodecParameters.ch_layout(channelLayout);
-				
+
 				audioCodecParameters.codec_id(AV_CODEC_ID_AAC);
 				audioCodecParameters.codec_type(AVMEDIA_TYPE_AUDIO);
 
@@ -624,7 +624,7 @@ public boolean prepare() throws Exception {
 			videoStreamIndex = streamIndex;
 			streamIndex++;
 		}
-		
+
 
 		AVCodecParameters parameters = getAudioCodecParameters();
 		if (parameters != null) {
@@ -691,7 +691,7 @@ public boolean prepareFromInputFormatContext(AVFormatContext inputFormatContext)
 			AVStream stream = inputFormatContext.streams(i);
 			AVCodecParameters codecpar = stream.codecpar();
 			if (codecpar.codec_type() == AVMEDIA_TYPE_VIDEO && !isBlacklistCodec(codecpar.codec_id())) {
-				
+
 				videoTimeBase = inputFormatContext.streams(i).time_base();
 				logger.info("Video format codec Id: {} width:{} height:{} for stream: {} source index:{} target index:{}", codecpar.codec_id(), codecpar.width(), codecpar.height(), streamId, i, streamIndex);
 				width = codecpar.width();
@@ -837,12 +837,12 @@ public void prepareMuxerIO()
 	 */
 	public void changeStreamQualityParameters(String streamId, String quality, double speed, int inputQueueSize) {
 		long now = System.currentTimeMillis();
-		
+
 		//increase updating time to 5 seconds because it may cause some issues in mongodb updates and no need to update every 5 seconds
 		if ((now - lastQualityUpdateTime) > 5000 &&
 				((quality != null && !quality.equals(oldQuality)) || oldspeed == 0 || Math.abs(speed - oldspeed) > 0.05)) 
 		{
-			
+
 			logger.info("Stream queue size:{} for streamId:{} ", inputQueueSize, streamId);
 			lastQualityUpdateTime = now;
 			getStreamHandler().setQualityParameters(streamId, quality, speed, inputQueueSize);
@@ -1003,17 +1003,17 @@ public void execute()
 					return;
 
 				}
-				
+
 				IStreamCodecInfo codecInfo = broadcastStream.getCodecInfo();
 				enableVideo = codecInfo.hasVideo();
 				enableAudio = codecInfo.hasAudio();
-				
+
 				getVideoDataConf(codecInfo);
 				getAudioDataConf(codecInfo);
-				
+
 				// Sometimes AAC Sequenece Header is received later 
 				// so that we check if we get the audio codec parameters correctly
-				
+
 				if (enableVideo && enableAudio && getAudioCodecParameters() != null)
 				{
 					logger.info("Video and audio is enabled in stream:{} queue size: {}", streamId, queueSize.get());
@@ -1236,8 +1236,8 @@ public void updateQualityParameters(long pts, AVRational timebase) {
 			}
 		}
 		changeStreamQualityParameters(this.streamId, null, speed, getInputQueueSize());
-		
-		
+
+
 	}
 
 	public void closeRtmpConnection() {
@@ -1319,9 +1319,6 @@ public synchronized void closeResources() {
 
 		writeTrailer();
 
-		isRecording.set(false);
-
-
 		if (videoExtraDataPointer != null) {
 			av_free(videoExtraDataPointer.position(0));
 			videoExtraDataPointer.close();
@@ -1336,9 +1333,11 @@ public synchronized void closeResources() {
 
 		changeStreamQualityParameters(this.streamId, null, 0, getInputQueueSize());
 		getStreamHandler().muxAdaptorRemoved(this);
+
+		isRecording.set(false);
 	}
-	
-	
+
+
 	/**
 	 * This method means that if the MuxAdaptor writes 
 	 * incoming packets to muxers({@link MuxAdaptor#muxerList}) directly without any StreamAdaptor/Encoders
@@ -1423,7 +1422,7 @@ public void stop(boolean shutdownCompletely) {
 	}
 
 	public int getInputQueueSize() {
-		return queueSize .get();
+		return queueSize.get();
 	}
 
 	public boolean isStopRequestExist() {
@@ -1444,7 +1443,7 @@ public void writeBufferedPacket()
 					while(!bufferQueue.isEmpty())
 					{
 						IStreamPacket tempPacket = peekTheNextPacketFromBuffer();
-						
+
 						long now = System.currentTimeMillis();
 						long pktTimeDifferenceMs = tempPacket.getTimestamp() - firstPacketReadyToSentTimeMs;
 						long passedTime = now - bufferingFinishTimeMs;
@@ -1493,7 +1492,7 @@ public IStreamPacket peekTheNextPacketFromBuffer() {
 				}	
 			}
 		}		
-		
+
 		return tempPacket;
 	}
 
@@ -1521,6 +1520,17 @@ public void packetReceived(IBroadcastStream stream, IStreamPacket packet)
 			logger.info("first received frame timestamp: {} for stream:{} ", lastFrameTimestamp, streamId);
 			firstReceivedFrameTimestamp = lastFrameTimestamp;
 		}
+		if (stopRequestExist) 
+		{
+			//there may be a bug that red5 cannot stop the stream ClientBroadcastStream and it may fill the memory
+			logger.warn("Stop request exist and dropping incoming packet for stream:{}", streamId);
+			//try to close the connection again
+			closeRtmpConnection();
+			return;
+		}
+
+
+
 		queueSize.incrementAndGet();
 
 		CachedEvent event = new CachedEvent();
@@ -1530,6 +1540,7 @@ public void packetReceived(IBroadcastStream stream, IStreamPacket packet)
 		event.setTimestamp(packet.getTimestamp());
 
 		streamPacketQueue.add(event);
+
 	}
 
 	@Override
@@ -1716,7 +1727,7 @@ public RecordMuxer startRecording(RecordType recordType, int resolutionHeight) {
 		if(recordType == RecordType.MP4) {
 			Mp4Muxer mp4Muxer = createMp4Muxer();
 			muxer = mp4Muxer;
-			
+
 			addMuxer(muxer, resolutionHeight);
 		}
 		else if(recordType == RecordType.WEBM) {
@@ -1732,7 +1743,7 @@ else if(recordType == RecordType.WEBM) {
 	public boolean prepareMuxer(Muxer muxer, int resolutionHeight) 
 	{
 		boolean streamAdded = false;
-		
+
 		muxer.init(scope, streamId, resolutionHeight, broadcast != null ? broadcast.getSubFolder(): null, 0);
 		logger.info("prepareMuxer for stream:{} muxer:{}", streamId, muxer.getClass().getSimpleName());
 
@@ -1774,7 +1785,7 @@ public boolean prepareMuxer(Muxer muxer, int resolutionHeight)
 		if (streamAdded) 
 		{
 			prepared = muxer.prepareIO();
-	
+
 			if (prepared) 
 			{
 				prepared = addMuxerInternal(muxer);
@@ -1784,7 +1795,7 @@ public boolean prepareMuxer(Muxer muxer, int resolutionHeight)
 			{
 				logger.warn("Muxer:{} cannot be prepared for streamId:{}", muxer.getClass().getSimpleName(), streamId);
 			}
-	
+
 			//TODO: Check to release the resources if it's not already released
 		}
 
@@ -1893,7 +1904,7 @@ public void endpointStatusHealthCheck(String url)
 		healthCheckPeriodMS = appSettings.getEndpointHealthCheckPeriodMs();
 		vertx.setPeriodic(healthCheckPeriodMS, id ->
 		{
-			
+
 			String status = statusMap.getValueOrDefault(url, null);
 			logger.info("Checking the endpoint health for: {} and status: {} ", url, status);
 			//Broadcast might get deleted in the process of checking
@@ -2202,7 +2213,7 @@ public AVRational getVideoTimeBase() {
 	public AVRational getAudioTimeBase() {
 		return audioTimeBase;
 	}
-	
+
 	public void setVideoTimeBase(AVRational videoTimeBase) {
 		this.videoTimeBase = videoTimeBase;
 	}
@@ -2237,7 +2248,7 @@ public void setIsRecording(boolean isRecording) {
 	public void setAudioDataConf(byte[] audioDataConf) {
 		this.audioDataConf = audioDataConf;
 	}
-	
+
 	public boolean isBlacklistCodec(int codecId) {
 		return (codecId == AV_CODEC_ID_PNG);
 	}