From df8afcb4283bc584776d3ae81fea96ee0464aaf7 Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Mon, 27 Jul 2026 23:00:37 +0100 Subject: [PATCH] cleaned up StreamActions interface --- .../ddf/vodsystem/services/MediaService.java | 9 +- .../services/media/StreamActionsService.java | 275 ++++++++++-------- .../media/StreamActionsServiceTest.java | 61 +++- 3 files changed, 215 insertions(+), 130 deletions(-) diff --git a/src/main/java/com/ddf/vodsystem/services/MediaService.java b/src/main/java/com/ddf/vodsystem/services/MediaService.java index 57deb71..712e0e0 100644 --- a/src/main/java/com/ddf/vodsystem/services/MediaService.java +++ b/src/main/java/com/ddf/vodsystem/services/MediaService.java @@ -134,10 +134,9 @@ public class MediaService { Job job = jobRegistryService.generateJob(); StreamActionsService.SegmentSelection ctx = resolveSegments(user, startTime, endTime); - float duration = (endTime.toEpochMilli() - startTime.toEpochMilli()) / 1000f; Path outputFile = directoryService.getVodsDir(user.getId()).resolve(UUID.randomUUID() + ".mp4"); - return dispatchSaveJob(job, ctx, duration, outputFile, () -> + return dispatchSaveJob(job, ctx, outputFile, () -> vodService.persist( title == null ? Instant.now().toString() : title, description == null ? "" : description, @@ -214,7 +213,7 @@ public class MediaService { StreamActionsService.SegmentSelection ctx = resolveSegments(user, startTime, endTime); Path outputFile = directoryService.getTempOutputDir().resolve(UUID.randomUUID() + ".mp4"); - return dispatchSaveJob(job, ctx, duration, outputFile, () -> + return dispatchSaveJob(job, ctx, outputFile, () -> clipService.persistClip( title == null ? Instant.now().toString() : title, description == null ? "" : description, @@ -230,9 +229,9 @@ public class MediaService { return streamActionsService.selectSegments(streamDir, startTime, endTime); } - private Job dispatchSaveJob(Job job, StreamActionsService.SegmentSelection ctx, float duration, Path outputFile, Runnable onSuccess) { + private Job dispatchSaveJob(Job job, StreamActionsService.SegmentSelection ctx, Path outputFile, Runnable onSuccess) { job.setState(JobState.PROCESSING); - streamActionsService.saveSection(ctx.segments(), ctx.trimOffset(), duration, outputFile, job.getProgressTracker()) + streamActionsService.saveSection(ctx, outputFile, job.getProgressTracker()) .thenRun(() -> { job.setDownload(outputFile); job.setState(JobState.SUCCEEDED); diff --git a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java index ed37183..553ce78 100644 --- a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java +++ b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java @@ -16,17 +16,16 @@ import java.util.Comparator; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.stream.Collectors; import java.util.stream.Stream; /** - * Service responsible for performing stream-related media actions, such as - * extracting and saving a trimmed section of video from a sequence of - * recorded segments. + * Extracts and saves a trimmed section of a stream from its recorded HLS + * segments. *

- * 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. + * The usual flow is {@link #selectSegments} to resolve the segments covering a + * time range into a {@link SegmentSelection}, then {@link #saveSection} to mux + * that selection to a file. Both are backed by {@code ffmpeg} and run on a + * dedicated executor so long-running encodes never block request threads. */ @Service public class StreamActionsService { @@ -40,66 +39,133 @@ public class StreamActionsService { } /** - * Concatenates a list of video segments and extracts a trimmed section - * from the result, writing the output to {@code outputFile}. - *

- * 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. + * A resolved section of a stream, ready to be muxed by {@link #saveSection}. * - * @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 the ffmpeg - * command output on success, or completes exceptionally on failure + * @param segments the segments to concatenate, in playback order; never empty + * @param trimOffset seconds to trim from the start of {@code segments.getFirst()} + * so playback begins at the requested moment, or {@code 0} + * when the first segment already starts there + * @param duration seconds of footage to keep, measured from the trim point. + * When the requested start fell in a recording gap and the + * leading segment was dropped, this is measured from the + * later real start so the section still ends where requested + */ + public record SegmentSelection(List segments, float trimOffset, float duration) {} + + /** + * Resolves the recorded {@code .ts} segments covering {@code [startTime, endTime)} + * into a {@link SegmentSelection} ready for {@link #saveSection}. + *

+ * The trim offset into the first segment is derived from {@code startTime}. If + * that start falls in a recording gap after the earliest overlapping segment + * (so the segment holds no footage for it), the segment is dropped and the + * selection instead begins at the next recorded footage. + * + * @param streamDirectory directory holding the stream's {@code .ts} segments + * @param startTime inclusive start of the requested section + * @param endTime exclusive end of the requested section; must be after {@code startTime} + * @return the segments, trim offset, and duration describing the section + * @throws IllegalArgumentException if {@code startTime} is not before {@code endTime}, + * or no segment holds footage for the range + * @throws IOException if listing the directory or probing a segment fails + */ + public SegmentSelection selectSegments(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { + List segments = findOverlappingSegments(streamDirectory, startTime, endTime); + if (segments.isEmpty()) { + throw new IllegalArgumentException("No stream segments found in the given time range"); + } + + long firstMs = parseTimestampMs(segments.getFirst()); + float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstMs) / 1000f); + + // A first segment shorter than the trim offset means the requested start + // lands in a gap after it; drop it and begin at the next recorded footage. + if (trimOffset > 0 && probeDurationSeconds(segments.getFirst()) < trimOffset) { + segments.removeFirst(); + if (segments.isEmpty()) { + throw new IllegalArgumentException("No recorded footage at the requested start time"); + } + firstMs = parseTimestampMs(segments.getFirst()); + trimOffset = 0f; + } + + // Measure duration from where playback actually begins so a dropped + // leading segment shortens the section rather than overrunning endTime. + long effectiveStartMs = firstMs + (long) (trimOffset * 1000); + float duration = (endTime.toEpochMilli() - effectiveStartMs) / 1000f; + + return new SegmentSelection(segments, trimOffset, duration); + } + + /** + * Muxes a {@link SegmentSelection} to {@code outputFile}, trimming the leading + * segment to {@code selection.trimOffset()} and keeping {@code selection.duration()} + * seconds of footage. + *

+ * Runs asynchronously; failures (process-launch errors, ffmpeg failures, + * interruption) complete the returned future exceptionally rather than being + * thrown, so callers should use {@link CompletableFuture#exceptionally} or + * {@link CompletableFuture#handle} rather than try/catch. + * + * @param selection the segments, trim offset, and duration to write + * (see {@link #selectSegments}) + * @param outputFile destination for the muxed section; overwritten if present + * @param progressTracker receives progress updates as ffmpeg runs + * @return a future completing with the ffmpeg output, or exceptionally on failure */ @Async("ffmpegTaskExecutor") public CompletableFuture saveSection( - List segments, - float trimOffset, - float duration, + SegmentSelection selection, Path outputFile, ProgressTracker progressTracker ) { + List segments = selection.segments(); + float trimOffset = selection.trimOffset(); + float duration = selection.duration(); if (segments.isEmpty()) { - return CompletableFuture.failedFuture(new IllegalArgumentException("Segment list cannot be empty")); + return CompletableFuture.failedFuture(new IllegalArgumentException("Selection must contain at least one segment")); } - if (trimOffset < 0 || duration <= 0) { return CompletableFuture.failedFuture(new IllegalArgumentException( "trimOffset must be >= 0 and duration must be > 0 (got " + trimOffset + ", " + duration + ")")); } - Path headSegment = null; + Path headFragment = null; + Path concatList = null; try { - // Unique per invocation so concurrent saves by different users never - // collide on the same head fragment; cleaned up in the finally block. - headSegment = Files.createTempFile("reencode-section-", ".ts"); - encodeHead(segments.getFirst(), trimOffset, headSegment); + // Only the first segment is trimmed, so re-encode just that fragment and + // stream-copy the rest. A zero offset means the first segment already + // starts at the requested point, so re-encoding it would be wasted work. + List fragments; + if (trimOffset > 0) { + headFragment = Files.createTempFile("reencode-section-", ".ts"); + encodeHead(segments.getFirst(), trimOffset, headFragment); - String concatInput = Stream.concat(Stream.of(headSegment), segments.stream().skip(1)) - .map(p -> p.toAbsolutePath().toString()) - .collect(Collectors.joining("|")); + fragments = new ArrayList<>(); + fragments.add(headFragment); + fragments.addAll(segments.subList(1, segments.size())); + } else { + fragments = segments; + } + + // The concat demuxer (unlike the concat: byte protocol) rebases each + // fragment's timestamps to run continuously across the joins, absorbing + // the discontinuity between the PTS-0 head and the copied fragments. + concatList = writeConcatList(fragments); List command = List.of( "ffmpeg", "-y", - "-i", "concat:" + concatInput, + "-f", "concat", + "-safe", "0", + "-i", concatList.toAbsolutePath().toString(), "-t", String.valueOf(duration), "-c", "copy", outputFile.toAbsolutePath().toString() ); - logger.info("Saving section with ({} segments) to '{}'", - segments.size(), outputFile); + logger.info("Saving section ({} segments) to '{}'", segments.size(), outputFile); CommandOutput output = commandRunner.run(command, line -> commandRunner.setProgress(line, progressTracker, duration)); progressTracker.markComplete(); @@ -115,25 +181,47 @@ public class StreamActionsService { logger.error("Failed to save section to '{}': {}", outputFile, e.toString()); return CompletableFuture.failedFuture(e); } finally { - if (headSegment != null) { - try { - Files.deleteIfExists(headSegment); - } catch (IOException e) { - logger.warn("Failed to delete temp head fragment '{}': {}", headSegment, e.toString()); - } - } + deleteTempQuietly(headFragment); + deleteTempQuietly(concatList); } } - private List getSegmentsInRange(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { + /** + * Parses the start timestamp encoded in an HLS segment's filename (e.g. + * {@code 1712345678.ts}). nginx names fragments in whole seconds, so + * second-scale values are scaled up to milliseconds. + * + * @param path a {@code .ts} segment path + * @return the segment's start time in epoch milliseconds + * @throws IllegalArgumentException if {@code path} is not a {@code .ts} file + * @throws NumberFormatException if the filename stem is not numeric + */ + public long parseTimestampMs(Path path) { + String fullName = path.getFileName().toString(); + + if (!fullName.endsWith(".ts")) { + throw new IllegalArgumentException("File must end in ts"); + } + + String name = fullName.substring(0, fullName.length() - 3); + long value = Long.parseLong(name); + return value < 1_000_000_000_000L ? value * 1000L : value; + } + + /** + * Lists the {@code .ts} segments whose playback interval overlaps + * {@code [startTime, endTime)}, in playback order. A segment spans from its + * own timestamp up to the next segment's timestamp (or open-ended for the last). + */ + private List findOverlappingSegments(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { long startMs = startTime.toEpochMilli(); long endMs = endTime.toEpochMilli(); - if (endTime.toEpochMilli() <= startTime.toEpochMilli()) { + if (endMs <= startMs) { throw new IllegalArgumentException("Start time must be before end time"); } - List results = new ArrayList<>(); + List overlapping = new ArrayList<>(); try (Stream files = Files.list(streamDirectory)) { List streamFiles = files @@ -151,54 +239,12 @@ public class StreamActionsService { : Long.MAX_VALUE; if (segmentStart < endMs && segmentEnd > startMs) { - results.add(streamFile); + overlapping.add(streamFile); } } } - return results; - } - - /** - * The segments backing a save and the offset into the first one at which - * the requested section begins. - * - * @param segments the ordered segments to concatenate, in playback order - * @param trimOffset the offset, in seconds, into {@code segments.getFirst()} - * at which the requested section starts - */ - public record SegmentSelection(List segments, float trimOffset) {} - - /** - * Selects the recorded {@code .ts} segments overlapping {@code [startTime, endTime)} - * and computes the trim offset into the first segment. - *

- * If {@code startTime} falls in a stream gap after the earliest overlapping segment - * (so that segment holds no footage for the requested start), the segment is dropped - * and the selection begins at the first genuinely-recorded footage instead. - * - * @throws IllegalArgumentException if no segments hold footage for the requested range - * @throws IOException if listing the stream directory or probing a segment fails - */ - public SegmentSelection selectSegments(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { - List segments = getSegmentsInRange(streamDirectory, startTime, endTime); - if (segments.isEmpty()) { - throw new IllegalArgumentException("No stream segments found in the given time range"); - } - - long firstMs = parseTimestampMs(segments.getFirst()); - float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstMs) / 1000f); - - // if the first segment is longer than the trim-offset - if (trimOffset > 0 && probeDurationSeconds(segments.getFirst()) < trimOffset) { - segments.removeFirst(); - if (segments.isEmpty()) { - throw new IllegalArgumentException("No recorded footage at the requested start time"); - } - trimOffset = 0f; - } - - return new SegmentSelection(segments, trimOffset); + return overlapping; } private float probeDurationSeconds(Path segment) throws IOException { @@ -212,24 +258,6 @@ public class StreamActionsService { } } - public long parseTimestampMs(Path path) { - String fullName = path.getFileName().toString(); - - if (!fullName.endsWith(".ts")) { - throw new IllegalArgumentException("File must end in ts"); - } - - String name = fullName.substring(0, fullName.length() - 3); - long value = Long.parseLong(name); - // nginx hls_fragment_naming system uses seconds; convert to ms - return value < 1_000_000_000_000L ? value * 1000L : value; - } - - /** - * Re-encodes {@code firstSegment} from {@code trimOffset} to its end, - * writing an MPEG-TS fragment that can be stream-copy concatenated with - * the untouched fragments that follow it. - */ private void encodeHead(Path firstSegment, float trimOffset, Path head) throws IOException, InterruptedException { @@ -251,11 +279,30 @@ public class StreamActionsService { ); logger.debug("Re-encoding head fragment '{}' from {}s", firstSegment, trimOffset); - commandRunner.run(command, line -> { /* head pass is short; not worth reporting */ }); + commandRunner.run(command, line -> { /* head pass is short; no progress reporting */ }); if (!Files.exists(head) || Files.size(head) == 0) { throw new IOException("Head re-encode produced no output — trimOffset " + trimOffset + "s may exceed the duration of " + firstSegment); } } + + private Path writeConcatList(List fragments) throws IOException { + Path listFile = Files.createTempFile("concat-list-", ".txt"); + List lines = fragments.stream() + .map(p -> "file '" + p.toAbsolutePath().toString().replace("'", "'\\''") + "'") + .toList(); + Files.write(listFile, lines); + return listFile; + } + + private void deleteTempQuietly(Path tempFile) { + if (tempFile != null) { + try { + Files.deleteIfExists(tempFile); + } catch (IOException e) { + logger.warn("Failed to delete temp file '{}': {}", tempFile, e.toString()); + } + } + } } diff --git a/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java index 8f91f84..eef3327 100644 --- a/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java +++ b/src/test/java/com/ddf/vodsystem/services/media/StreamActionsServiceTest.java @@ -247,6 +247,8 @@ class StreamActionsServiceTest { assertThat(selection.segments()).extracting(p -> p.getFileName().toString()) .containsExactly("30.ts", "33.ts"); assertThat(selection.trimOffset()).isEqualTo(1.5f); + // effective start equals the requested 31.5s, so duration runs to endTime: 40 - 31.5 + assertThat(selection.duration()).isEqualTo(8.5f); } @Test @@ -262,6 +264,10 @@ class StreamActionsServiceTest { assertThat(selection.segments()).extracting(p -> p.getFileName().toString()) .containsExactly("100.ts", "110.ts"); assertThat(selection.trimOffset()).isZero(); + // the dropped head shifts the effective start to 100s, so the duration + // shrinks to end at endTime (111 - 100 = 11s) instead of the 41s the + // original 70s->111s window would have implied + assertThat(selection.duration()).isEqualTo(11f); } @Test @@ -351,13 +357,15 @@ class StreamActionsServiceTest { stubRunnerProducingHead(commandOutput); - var future = service.saveSection(List.of(segmentA, segmentB), 5.0f, 12.0f, output, progress); + var future = service.saveSection( + new StreamActionsService.SegmentSelection(List.of(segmentA, segmentB), 5.0f, 12.0f), output, progress); assertThat(future.get()).isSameAs(commandOutput); assertThat(progress.isComplete()).isTrue(); - // saveSection issues two ffmpeg passes: the head re-encode (trims the - // first segment via -ss) followed by the stream-copy concat (-t/-c copy). + // With a non-zero trim offset saveSection issues two ffmpeg passes: the + // head re-encode (trims the first segment via -ss) followed by the + // concat-demuxer stream copy (-f concat / -t / -c copy). @SuppressWarnings("unchecked") ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); Mockito.verify(commandRunner, Mockito.times(2)).run(captor.capture(), any()); @@ -365,20 +373,50 @@ class StreamActionsServiceTest { List concatCommand = captor.getAllValues().get(1); assertThat(headCommand).containsSequence("-ss", "5.0"); + assertThat(concatCommand).containsSequence("-f", "concat"); assertThat(concatCommand).containsSequence("-t", "12.0"); assertThat(concatCommand).containsSequence("-c", "copy"); assertThat(concatCommand).endsWith(output.toAbsolutePath().toString()); } + @Test + void saveSection_zeroTrimOffset_skipsHeadReencodeAndRunsSingleConcatPass() throws Exception { + Path output = Path.of("/tmp/out/clip.mp4"); + ProgressTracker progress = new ProgressTracker(); + CommandOutput commandOutput = new CommandOutput(); + + when(commandRunner.run(anyList(), any())).thenReturn(commandOutput); + + var future = service.saveSection( + new StreamActionsService.SegmentSelection( + List.of(Path.of("/tmp/a.ts"), Path.of("/tmp/b.ts")), 0f, 10f), output, progress); + + assertThat(future.get()).isSameAs(commandOutput); + assertThat(progress.isComplete()).isTrue(); + + // trimOffset 0 means the first segment already starts at the requested point: + // no head re-encode pass, just a single concat-demuxer copy (no -ss). + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); + Mockito.verify(commandRunner, Mockito.times(1)).run(captor.capture(), any()); + List command = captor.getValue(); + + assertThat(command).containsSequence("-f", "concat"); + assertThat(command).containsSequence("-t", "10.0"); + assertThat(command).doesNotContain("-ss"); + } + @Test void saveSection_commandRunnerThrowsIOException_completesExceptionallyWithoutMarkingProgressComplete() throws Exception { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); - // The head re-encode is the first ffmpeg pass; its failure must surface. + // With a non-zero trim offset the head re-encode is the first ffmpeg pass; + // its failure must surface. when(commandRunner.run(anyList(), any())).thenThrow(new IOException("ffmpeg boom")); - var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, output, progress); + var future = service.saveSection( + new StreamActionsService.SegmentSelection(List.of(Path.of("/tmp/a.ts")), 2f, 10f), output, progress); assertThatThrownBy(future::get) .isInstanceOf(ExecutionException.class) @@ -391,7 +429,8 @@ class StreamActionsServiceTest { Path output = Path.of("/tmp/out/clip.mp4"); ProgressTracker progress = new ProgressTracker(); - var future = service.saveSection(List.of(), 0f, 10f, output, progress); + var future = service.saveSection( + new StreamActionsService.SegmentSelection(List.of(), 0f, 10f), output, progress); assertThatThrownBy(future::get) .isInstanceOf(ExecutionException.class) @@ -400,13 +439,13 @@ class StreamActionsServiceTest { } @Test - void saveSection_nullOutputFile_completesExceptionally() throws Exception { + void saveSection_nullOutputFile_completesExceptionally() { ProgressTracker progress = new ProgressTracker(); - // Let the head pass succeed so execution reaches the null output file. - stubRunnerProducingHead(new CommandOutput()); - - var future = service.saveSection(List.of(Path.of("/tmp/a.ts")), 0f, 10f, null, progress); + // trimOffset 0 skips the head pass; building the concat command dereferences + // the null output path, so no ffmpeg pass ever runs. + var future = service.saveSection( + new StreamActionsService.SegmentSelection(List.of(Path.of("/tmp/a.ts")), 0f, 10f), null, progress); assertThatThrownBy(future::get) .isInstanceOf(ExecutionException.class)