Merge pull request #6613 from ant-media/improve-vod-api

Improve VoD Upload API

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

@@ -1017,9 +1017,9 @@ public void notifyNoActiveSubtracksLeftInMainTrack(Broadcast mainTrack)
 		}
 	}
 	public void runScript(String scriptFile) {
-		vertx.executeBlocking(future -> {
+		vertx.executeBlocking(() -> {
 			try {
-				logger.info("running muxer finish script: {}", scriptFile);
+				logger.info("running script: {}", scriptFile);
 				Process exec = Runtime.getRuntime().exec(scriptFile);
 				int result = exec.waitFor();
 
@@ -1030,9 +1030,10 @@ public void runScript(String scriptFile) {
 				logger.error(ExceptionUtils.getStackTrace(e));
 				Thread.currentThread().interrupt();
 			}
-			future.complete();
+
+			return null;
 
-		}, null);
+		}, false);
 	}
 
 	public static String getRelativePath(String filePath){

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

@@ -175,6 +175,8 @@ public long getBroadcastCount(Map<String,String> broadcastMap) {
 	public abstract boolean delete(String id);
 
 	public abstract boolean deleteVod(String id);
+
+	public abstract boolean updateVoDProcessStatus(String id, String status);
 
 	/**
 	 * Returns the Broadcast List in order

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

@@ -112,7 +112,7 @@ else if(status.equals(IAntMediaStreamHandler.BROADCAST_STATUS_FINISHED)) {
 		}
 		return result;
 	}
-
+	
 	@Override
 	public boolean updateDuration(String id, long duration) {
 		Broadcast broadcast = broadcastMap.get(id);
@@ -296,6 +296,24 @@ public String addVod(VoD vod) {
 		}
 		return id;
 	}
+
+	@Override
+	public boolean updateVoDProcessStatus(String id, String status) {
+		VoD vod = vodMap.get(id);
+		if (vod != null) {
+			vod.setProcessStatus(status);
+			if (VoD.PROCESS_STATUS_PROCESSING.equals(status)) {
+				vod.setProcessStartTime(System.currentTimeMillis());
+			}
+			else if (VoD.PROCESS_STATUS_FAILED.equals(status) || VoD.PROCESS_STATUS_FINISHED.equals(status)) {
+				vod.setProcessEndTime(System.currentTimeMillis());
+			}
+			vodMap.put(id, vod);
+			return true;
+		}
+
+		return false;
+	}
 
 	@Override
 	public List<VoD> getVodList(int offset, int size, String sortBy, String orderBy, String filterStreamId, String search)

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

@@ -115,6 +115,28 @@ public boolean updateStatus(String id, String status) {
 		}
 		return result;
 	}
+
+	@Override
+	public boolean updateVoDProcessStatus(String id, String status) {
+		boolean result = false;
+		synchronized (this) {
+			String vodString = vodMap.get(id);
+			if (vodString != null) {
+				VoD vod = gson.fromJson(vodString, VoD.class);
+				if (VoD.PROCESS_STATUS_PROCESSING.equals(status)) {
+					vod.setProcessStartTime(System.currentTimeMillis());
+				}
+				else if (VoD.PROCESS_STATUS_FAILED.equals(status) || VoD.PROCESS_STATUS_FINISHED.equals(status)) {
+					vod.setProcessEndTime(System.currentTimeMillis());
+				}
+				vod.setProcessStatus(status);
+				vodMap.put(id, gson.toJson(vod));
+				result = true;
+			}
+		}
+
+		return result;
+	}
 
 	@Override
 	public boolean updateDuration(String id, long duration) {

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

@@ -266,6 +266,30 @@ else if(status.equals(IAntMediaStreamHandler.BROADCAST_STATUS_FINISHED))
 		return false;
 
 	}
+
+	@Override
+	public boolean updateVoDProcessStatus(String id, String status) {
+		synchronized (this) {
+			try {
+				Query<VoD> query = vodDatastore.find(VoD.class).filter(Filters.eq(VOD_ID, id));
+				Update<VoD> ops = query.update(set("processStatus", status));
+				if (VoD.PROCESS_STATUS_PROCESSING.equals(status)) 
+				{
+					ops.add(set("processStartTime", System.currentTimeMillis()));
+				}
+				else if (VoD.PROCESS_STATUS_FAILED.equals(status) || VoD.PROCESS_STATUS_FINISHED.equals(status)) 
+				{
+					ops.add(set("processEndTime", System.currentTimeMillis()));
+				}
+
+				UpdateResult update = ops.execute();
+				return update.getMatchedCount() == 1;
+			} catch (Exception e) {
+				logger.error(ExceptionUtils.getStackTrace(e));
+			}
+		}
+		return false;
+	}
 
 	/*
 	 * (non-Javadoc)

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

@@ -14,6 +14,7 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 
 import io.swagger.v3.oas.annotations.media.Schema;
+import io.swagger.v3.oas.annotations.media.Schema.AccessMode;
 
 
 @Entity("vod")
@@ -39,6 +40,16 @@ public class VoD implements Serializable {
 	 */
 	public static final String UPLOADED_VOD = "uploadedVod";
 
+
+	public static final String PROCESS_STATUS_INQUEUE = "inqueue";
+
+	public static final String PROCESS_STATUS_PROCESSING = "processing";
+
+	public static final String PROCESS_STATUS_FINISHED = "finished";
+
+	public static final String PROCESS_STATUS_FAILED = "failed";
+
+
 	@JsonIgnore
 	@Id
     private ObjectId dbId;
@@ -75,6 +86,16 @@ public class VoD implements Serializable {
 
     @Schema(description = "The file path for the preview of the VoD")
     private String previewFilePath;
+
+    @Schema(description = "The status of the VoD processing. It can be inqueue, processing, finished, failed")
+    private String processStatus;
+
+    @Schema(description = "The start time of the VoD processing in milliseconds (UTC- Unix epoch)", accessMode = AccessMode.READ_ONLY)
+    private long processStartTime;
+
+    @Schema(description = "The end time of the VoD processing in milliseconds", accessMode = AccessMode.READ_ONLY)
+    private long processEndTime;
+
 
 	public VoD() {
 		//default constructor is used to return not found vod in rest service 
@@ -186,4 +207,46 @@ public void setPreviewFilePath(String previewFilePath) {
 		this.previewFilePath = previewFilePath;
 	}
 
+	/**
+	 * @return the processStatus
+	 */
+	public String getProcessStatus() {
+		return processStatus;
+	}
+
+	/**
+	 * @param processStatus the processStatus to set
+	 */
+	public void setProcessStatus(String processStatus) {
+		this.processStatus = processStatus;
+	}
+
+	/**
+	 * @return the processStartTime
+	 */
+	public long getProcessStartTime() {
+		return processStartTime;
+	}
+
+	/**
+	 * @param processStartTime the processStartTime to set
+	 */
+	public void setProcessStartTime(long processStartTime) {
+		this.processStartTime = processStartTime;
+	}
+
+	/**
+	 * @return the processEndTime
+	 */
+	public long getProcessEndTime() {
+		return processEndTime;
+	}
+
+	/**
+	 * @param processEndTime the processEndTime to set
+	 */
+	public void setProcessEndTime(long processEndTime) {
+		this.processEndTime = processEndTime;
+	}
+
 }