refactor(media)!: unify processing under asynchronous job workflow #3
@@ -173,25 +173,26 @@ public class ClipService {
|
|||||||
return clipRepository.save(clip);
|
return clipRepository.save(clip);
|
||||||
}
|
}
|
||||||
|
|
||||||
public void persistClip(ClipOptions clipOptions,
|
public void persistClip(String title,
|
||||||
User user,
|
String description,
|
||||||
File tempFile,
|
User user,
|
||||||
String fileName) {
|
File clipFile,
|
||||||
File clipFile;
|
String fileName) {
|
||||||
|
File newClipFile;
|
||||||
File thumbnailFile;
|
File thumbnailFile;
|
||||||
|
|
||||||
// Move temp file from temp dir to output dir
|
// Move temp file from temp dir to output dir
|
||||||
try {
|
try {
|
||||||
clipFile = directoryService.getUserClipsFile(user.getId(), fileName);
|
newClipFile = directoryService.getUserClipsFile(user.getId(), fileName);
|
||||||
thumbnailFile = directoryService.getUserThumbnailsFile(user.getId(), fileName + ".png");
|
thumbnailFile = directoryService.getUserThumbnailsFile(user.getId(), fileName + ".png");
|
||||||
directoryService.cutFile(tempFile, clipFile);
|
directoryService.cutFile(clipFile, newClipFile);
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
throw new StorageException("Failed to move clip from temporary directory to output directory", e);
|
throw new StorageException("Failed to move clip from temporary directory to output directory", e);
|
||||||
}
|
}
|
||||||
|
|
||||||
ClipOptions clipMetadata;
|
ClipOptions clipMetadata;
|
||||||
try {
|
try {
|
||||||
clipMetadata = metadataService.getVideoMetadata(clipFile).get();
|
clipMetadata = metadataService.getVideoMetadata(newClipFile).get();
|
||||||
} catch (InterruptedException | ExecutionException e) {
|
} catch (InterruptedException | ExecutionException e) {
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
throw new FFMPEGException("Error retrieving video metadata for clip: " + e.getMessage());
|
throw new FFMPEGException("Error retrieving video metadata for clip: " + e.getMessage());
|
||||||
@@ -199,7 +200,7 @@ public class ClipService {
|
|||||||
|
|
||||||
// Thumbnail generation can fail with error propagation
|
// Thumbnail generation can fail with error propagation
|
||||||
try {
|
try {
|
||||||
thumbnailService.createThumbnail(clipFile, thumbnailFile, 0.0f);
|
thumbnailService.createThumbnail(newClipFile, thumbnailFile, 0.0f);
|
||||||
} catch (InterruptedException e) {
|
} catch (InterruptedException e) {
|
||||||
logger.error("Thumbnail generation interrupted for user: {}", user.getId(), e);
|
logger.error("Thumbnail generation interrupted for user: {}", user.getId(), e);
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
@@ -207,10 +208,10 @@ public class ClipService {
|
|||||||
logger.error("Error generating thumbnail for user: {}", user.getId(), e);
|
logger.error("Error generating thumbnail for user: {}", user.getId(), e);
|
||||||
}
|
}
|
||||||
|
|
||||||
clipMetadata.setTitle(clipOptions.getTitle());
|
clipMetadata.setTitle(title);
|
||||||
clipMetadata.setDescription(clipOptions.getDescription());
|
clipMetadata.setDescription(description);
|
||||||
|
|
||||||
Clip clip = saveClip(clipMetadata, user, clipFile.getAbsolutePath(), thumbnailFile.getAbsolutePath());
|
Clip clip = saveClip(clipMetadata, user, newClipFile.getAbsolutePath(), thumbnailFile.getAbsolutePath());
|
||||||
logger.info("Clip created successfully with ID: {}", clip.getId());
|
logger.info("Clip created successfully with ID: {}", clip.getId());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,7 +5,6 @@ import java.io.IOException;
|
|||||||
import java.nio.file.Files;
|
import java.nio.file.Files;
|
||||||
import java.nio.file.Paths;
|
import java.nio.file.Paths;
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
|
||||||
|
|
||||||
import com.ddf.vodsystem.dto.Job;
|
import com.ddf.vodsystem.dto.Job;
|
||||||
import com.ddf.vodsystem.entities.User;
|
import com.ddf.vodsystem.entities.User;
|
||||||
@@ -133,7 +132,8 @@ public class JobOrchestrationService {
|
|||||||
compressionService.compress(job.getInputFile(), job.getOutputFile(), job.getOutputClipOptions(), job.getStatus().getProcess())
|
compressionService.compress(job.getInputFile(), job.getOutputFile(), job.getOutputClipOptions(), job.getStatus().getProcess())
|
||||||
.thenRun(() -> user.ifPresent(userVal ->
|
.thenRun(() -> user.ifPresent(userVal ->
|
||||||
clipService.persistClip(
|
clipService.persistClip(
|
||||||
job.getOutputClipOptions(),
|
job.getOutputClipOptions().getTitle(),
|
||||||
|
job.getOutputClipOptions().getDescription(),
|
||||||
userVal,
|
userVal,
|
||||||
job.getOutputFile(),
|
job.getOutputFile(),
|
||||||
job.getInputFile().getName()
|
job.getInputFile().getName()
|
||||||
|
|||||||
@@ -33,15 +33,18 @@ public class StreamService {
|
|||||||
private final UserService userService;
|
private final UserService userService;
|
||||||
private final StreamActionsService streamActionsService;
|
private final StreamActionsService streamActionsService;
|
||||||
private final DirectoryService directoryService;
|
private final DirectoryService directoryService;
|
||||||
|
private final ClipService clipService;
|
||||||
|
|
||||||
public StreamService(StreamRepository streamRepository,
|
public StreamService(StreamRepository streamRepository,
|
||||||
UserService userService,
|
UserService userService,
|
||||||
StreamActionsService streamActionsService,
|
StreamActionsService streamActionsService,
|
||||||
DirectoryService directoryService) {
|
DirectoryService directoryService,
|
||||||
|
ClipService clipService) {
|
||||||
this.streamRepository = streamRepository;
|
this.streamRepository = streamRepository;
|
||||||
this.userService = userService;
|
this.userService = userService;
|
||||||
this.streamActionsService = streamActionsService;
|
this.streamActionsService = streamActionsService;
|
||||||
this.directoryService = directoryService;
|
this.directoryService = directoryService;
|
||||||
|
this.clipService = clipService;
|
||||||
}
|
}
|
||||||
|
|
||||||
public Stream startStream(String streamKey) {
|
public Stream startStream(String streamKey) {
|
||||||
@@ -121,9 +124,8 @@ public class StreamService {
|
|||||||
* or if no stream segments exist in the given range
|
* or if no stream segments exist in the given range
|
||||||
* @throws NotAuthenticated if no user is currently authenticated
|
* @throws NotAuthenticated if no user is currently authenticated
|
||||||
* @throws IOException if reading the stream directory or its segments fails
|
* @throws IOException if reading the stream directory or its segments fails
|
||||||
* @throws InterruptedException if the thread is interrupted while performing the save
|
|
||||||
*/
|
*/
|
||||||
public void saveSection(Instant startTime, Instant endTime) throws IOException, InterruptedException {
|
public void saveSection(Instant startTime, Instant endTime) throws IOException {
|
||||||
if (startTime.isAfter(endTime)) {
|
if (startTime.isAfter(endTime)) {
|
||||||
throw new IllegalArgumentException("Start time must be before end time");
|
throw new IllegalArgumentException("Start time must be before end time");
|
||||||
}
|
}
|
||||||
@@ -159,9 +161,8 @@ public class StreamService {
|
|||||||
* or if no stream segments exist in the computed window
|
* or if no stream segments exist in the computed window
|
||||||
* @throws NotAuthenticated if no user is currently authenticated
|
* @throws NotAuthenticated if no user is currently authenticated
|
||||||
* @throws IOException if reading the stream directory or its segments fails
|
* @throws IOException if reading the stream directory or its segments fails
|
||||||
* @throws InterruptedException if the thread is interrupted while performing the save
|
|
||||||
*/
|
*/
|
||||||
public void clip(float duration) throws IOException, InterruptedException {
|
public void clip(float duration) throws IOException {
|
||||||
if (duration <= 0 || duration > CLIP_MAX_LENGTH) {
|
if (duration <= 0 || duration > CLIP_MAX_LENGTH) {
|
||||||
throw new IllegalArgumentException("Clip length must be between 0 and " + CLIP_MAX_LENGTH + " seconds");
|
throw new IllegalArgumentException("Clip length must be between 0 and " + CLIP_MAX_LENGTH + " seconds");
|
||||||
}
|
}
|
||||||
@@ -170,7 +171,30 @@ public class StreamService {
|
|||||||
Instant endTime = Instant.now();
|
Instant endTime = Instant.now();
|
||||||
Instant startTime = endTime.minus(Duration.ofMillis((long) (duration * 1000)));
|
Instant startTime = endTime.minus(Duration.ofMillis((long) (duration * 1000)));
|
||||||
|
|
||||||
saveSection(startTime, endTime);
|
User user = userService.getLoggedInUser()
|
||||||
|
.orElseThrow(() -> new NotAuthenticated("User is not authenticated"));
|
||||||
|
|
||||||
|
File streamDirectory = directoryService.getStreamDir(user.getStreamKey());
|
||||||
|
List<File> fileSegments = getSegmentsInRange(streamDirectory.getAbsolutePath(), startTime, endTime);
|
||||||
|
|
||||||
|
if (fileSegments.isEmpty()) {
|
||||||
|
throw new IllegalArgumentException("No stream segments found in the given time range");
|
||||||
|
}
|
||||||
|
|
||||||
|
long firstSegmentMs = parseTimestampMs(fileSegments.getFirst());
|
||||||
|
float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstSegmentMs) / 1000f);
|
||||||
|
|
||||||
|
File outputFile = directoryService.getTempOutputFile(UUID.randomUUID() + ".mp4");
|
||||||
|
streamActionsService.saveSection(fileSegments, trimOffset, duration, outputFile)
|
||||||
|
.thenRun(() ->
|
||||||
|
clipService.persistClip(
|
||||||
|
Instant.now().toString(),
|
||||||
|
"",
|
||||||
|
user,
|
||||||
|
outputFile,
|
||||||
|
outputFile.getName()
|
||||||
|
)
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Scheduled(fixedDelay = 30_000)
|
@Scheduled(fixedDelay = 30_000)
|
||||||
|
|||||||
@@ -1,54 +1,80 @@
|
|||||||
package com.ddf.vodsystem.services.media;
|
package com.ddf.vodsystem.services.media;
|
||||||
|
|
||||||
import com.ddf.vodsystem.services.DirectoryService;
|
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
import org.springframework.scheduling.annotation.Async;
|
import org.springframework.scheduling.annotation.Async;
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
import java.io.File;
|
import java.io.File;
|
||||||
import java.io.IOException;
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.UUID;
|
|
||||||
import java.util.concurrent.CompletableFuture;
|
import java.util.concurrent.CompletableFuture;
|
||||||
import java.util.stream.Collectors;
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Service responsible for performing stream-related media actions, such as
|
||||||
|
* extracting and saving a trimmed section of video from a sequence of
|
||||||
|
* recorded segments.
|
||||||
|
* <p>
|
||||||
|
* Operations in this service are backed by {@code ffmpeg} and run
|
||||||
|
* asynchronously on a dedicated executor so that potentially long-running
|
||||||
|
* encoding/concatenation work does not block request-handling threads.
|
||||||
|
*/
|
||||||
@Service
|
@Service
|
||||||
public class StreamActionsService {
|
public class StreamActionsService {
|
||||||
private static final Logger logger = LoggerFactory.getLogger(StreamActionsService.class);
|
private static final Logger logger = LoggerFactory.getLogger(StreamActionsService.class);
|
||||||
|
|
||||||
private final DirectoryService directoryService;
|
/**
|
||||||
|
* Concatenates a list of video segments and extracts a trimmed section
|
||||||
public StreamActionsService(DirectoryService directoryService) {
|
* from the result, writing the output to {@code outputFile}.
|
||||||
this.directoryService = directoryService;
|
* <p>
|
||||||
}
|
* Failures (process launch errors, ffmpeg failures, interruption) are
|
||||||
|
* surfaced via the returned future rather than thrown, in line with
|
||||||
|
* normal {@link CompletableFuture} conventions — callers should use
|
||||||
|
* {@code .exceptionally()} / {@code .handle()} rather than a try/catch.
|
||||||
|
*
|
||||||
|
* @param segments the ordered list of segment files to concatenate,
|
||||||
|
* in playback order
|
||||||
|
* @param trimOffset the offset, in seconds, from the start of the
|
||||||
|
* concatenated stream at which to begin the trimmed
|
||||||
|
* section
|
||||||
|
* @param duration the duration, in seconds, of the trimmed section
|
||||||
|
* to extract
|
||||||
|
* @param outputFile the destination file the trimmed section will be
|
||||||
|
* written to; overwritten if it already exists
|
||||||
|
* @return a {@link CompletableFuture} that completes with
|
||||||
|
* {@code outputFile} on success, or completes exceptionally on
|
||||||
|
* failure
|
||||||
|
*/
|
||||||
@Async("ffmpegTaskExecutor")
|
@Async("ffmpegTaskExecutor")
|
||||||
public CompletableFuture<File> saveSection(
|
public CompletableFuture<File> saveSection(
|
||||||
List<File> segments,
|
List<File> segments,
|
||||||
float trimOffset,
|
float trimOffset,
|
||||||
float duration,
|
float duration,
|
||||||
File outputFile
|
File outputFile
|
||||||
) throws IOException, InterruptedException {
|
) {
|
||||||
String concatInput = segments.stream()
|
try {
|
||||||
.map(File::getAbsolutePath)
|
String concatInput = segments.stream()
|
||||||
.collect(Collectors.joining("|"));
|
.map(File::getAbsolutePath)
|
||||||
|
.collect(Collectors.joining("|"));
|
||||||
|
|
||||||
List<String> command = List.of(
|
List<String> command = List.of(
|
||||||
"ffmpeg",
|
"ffmpeg",
|
||||||
"-y",
|
"-y",
|
||||||
"-i", "concat:" + concatInput,
|
"-i", "concat:" + concatInput,
|
||||||
"-ss", String.valueOf(trimOffset),
|
"-ss", String.valueOf(trimOffset),
|
||||||
"-t", String.valueOf(duration),
|
"-t", String.valueOf(duration),
|
||||||
"-c", "copy",
|
"-c", "copy",
|
||||||
outputFile.getAbsolutePath()
|
outputFile.getAbsolutePath()
|
||||||
);
|
);
|
||||||
|
|
||||||
logger.info("Saving section with ({} segments) to '{}'",
|
logger.info("Saving section with ({} segments) to '{}'",
|
||||||
segments.size(), outputFile.getAbsolutePath());
|
segments.size(), outputFile.getAbsolutePath());
|
||||||
CommandRunner.run(command, line -> logger.debug("{}", line));
|
CommandRunner.run(command, line -> logger.debug("{}", line));
|
||||||
|
|
||||||
return CompletableFuture.completedFuture(outputFile);
|
return CompletableFuture.completedFuture(outputFile);
|
||||||
|
} catch (Exception e) {
|
||||||
|
logger.error("Failed to save section to '{}'", outputFile.getAbsolutePath(), e);
|
||||||
|
return CompletableFuture.failedFuture(e);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
Reference in New Issue
Block a user