davebluke
/
ctbrec
forked from j62/ctbrec
1
0
Fork 0
Code Pull Requests Activity

Use FFmpeg to merge HLS segments

This commit is contained in:
0xboobface 2019-12-07 16:58:56 +01:00
parent 32d7a788e0
commit 034fe81081
3 changed files with 67 additions and 523 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
client/src/main/java/ctbrec/ui/RecordingsTab.java
View File

@ -4,7 +4,6 @@ import static ctbrec.Recording.State.*;
import static javafx.scene.control.ButtonType.*;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URL;
import java.nio.file.NoSuchFileException;
@ -479,22 +478,23 @@ public class RecordingsTab extends Tab implements TabSelectionListener {
Thread t = new Thread(() -> {
try {
MergedHlsDownload download = new MergedHlsDownload(CamrecApplication.httpClient);
download.start(url.toString(), target, progress -> Platform.runLater(() -> {
if (progress == 100) {
recording.setStatus(FINISHED);
recording.setProgress(-1);
LOG.debug("Download finished for recording {}", recording.getPath());
} else {
recording.setStatus(DOWNLOADING);
recording.setProgress(progress);
}
}));
} catch (FileNotFoundException e) {
showErrorDialog(ERROR_WHILE_DOWNLOADING_RECORDING, "The target file couldn't be created", e);
LOG.error(ERROR_WHILE_DOWNLOADING_RECORDING, e);
} catch (IOException e) {
showErrorDialog(ERROR_WHILE_DOWNLOADING_RECORDING, "The recording could not be downloaded", e);
LOG.error(ERROR_WHILE_DOWNLOADING_RECORDING, e);
LOG.info("Downloading {}", url);
// download.start(url.toString(), target, progress -> Platform.runLater(() -> {
// if (progress == 100) {
// recording.setStatus(FINISHED);
// recording.setProgress(-1);
// LOG.debug("Download finished for recording {}", recording.getPath());
// } else {
// recording.setStatus(DOWNLOADING);
// recording.setProgress(progress);
// }
// }));
// } catch (FileNotFoundException e) {
// showErrorDialog(ERROR_WHILE_DOWNLOADING_RECORDING, "The target file couldn't be created", e);
// LOG.error(ERROR_WHILE_DOWNLOADING_RECORDING, e);
// } catch (IOException e) {
// showErrorDialog(ERROR_WHILE_DOWNLOADING_RECORDING, "The recording could not be downloaded", e);
// LOG.error(ERROR_WHILE_DOWNLOADING_RECORDING, e);
} finally {
Platform.runLater(() -> {
recording.setStatus(FINISHED);

35
common/src/main/java/ctbrec/recorder/download/hls/HlsDownload.java
View File

@ -14,6 +14,7 @@ import java.nio.file.Path;
import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
@ -67,6 +68,7 @@ public class HlsDownload extends AbstractHlsDownload {
this.config = config;
super.model = model;
DateTimeFormatter formatter = DateTimeFormatter.ofPattern(Config.RECORDING_DATE_FORMAT);
this.startTime = Instant.now();
String startTime = formatter.format(ZonedDateTime.ofInstant(this.startTime, ZoneId.systemDefault()));
Path modelDir = FileSystems.getDefault().getPath(config.getSettings().recordingsDir, model.getSanitizedNamed());
downloadDir = FileSystems.getDefault().getPath(modelDir.toString(), startTime);
@ -94,7 +96,6 @@ public class HlsDownload extends AbstractHlsDownload {
SegmentPlaylist playlist = getNextSegments(segments);
emptyPlaylistCheck(playlist);
if(nextSegmentNumber > 0 && playlist.seq > nextSegmentNumber) {
// TODO switch to a lower bitrate/resolution ?!?
waitFactor *= 2;
LOG.warn("Missed segments {} < {} in download for {} - setting wait factor to 1/{}", nextSegmentNumber, playlist.seq, model, waitFactor);
}
@ -116,25 +117,18 @@ public class HlsDownload extends AbstractHlsDownload {
break;
}
long wait = 0;
long waitForMillis = 0;
if(lastSegmentNumber == playlist.seq) {
// playlist didn't change -> wait for at least half the target duration
wait = (long) playlist.targetDuration * 1000 / waitFactor;
LOG.trace("Playlist didn't change... waiting for {}ms", wait);
waitForMillis = (long) playlist.targetDuration * 1000 / waitFactor;
LOG.trace("Playlist didn't change... waiting for {}ms", waitForMillis);
} else {
// playlist did change -> wait for at least last segment duration
wait = 1;
LOG.trace("Playlist changed... waiting for {}ms", wait);
waitForMillis = 1;
LOG.trace("Playlist changed... waiting for {}ms", waitForMillis);
}
try {
Thread.sleep(wait);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
if(running) {
LOG.error("Couldn't sleep between segment downloads. This might mess up the download!");
}
}
waitSomeTime(waitForMillis);
// this if check makes sure, that we don't decrease nextSegment. for some reason
// streamate playlists sometimes jump back. e.g. max sequence = 79 -> 80 -> 79
@ -156,10 +150,10 @@ public class HlsDownload extends AbstractHlsDownload {
} catch(HttpException e) {
if(e.getResponseCode() == 404) {
LOG.debug("Playlist not found (404). Model {} probably went offline", model);
waitSomeTime();
waitSomeTime(10_000);
} else if(e.getResponseCode() == 403) {
LOG.debug("Playlist access forbidden (403). Model {} probably went private or offline", model);
waitSomeTime();
waitSomeTime(10_000);
} else {
throw e;
}
@ -188,7 +182,7 @@ public class HlsDownload extends AbstractHlsDownload {
super.postprocess(recording);
}
private void generatePlaylist(Recording recording) {
protected void generatePlaylist(Recording recording) {
File recDir = recording.getAbsoluteFile();
if(!config.getSettings().generatePlaylist) {
return;
@ -356,11 +350,14 @@ public class HlsDownload extends AbstractHlsDownload {
* This is used to slow down retries, if something is wrong with the playlist.
* E.g. HTTP 403 or 404
*/
private void waitSomeTime() {
private void waitSomeTime(long waitForMillis) {
try {
Thread.sleep(10_000);
Thread.sleep(waitForMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
if(running) {
LOG.error("Couldn't sleep. This might mess up the download!");
}
}
}
}

521
common/src/main/java/ctbrec/recorder/download/hls/MergedHlsDownload.java
View File

@ -1,516 +1,63 @@
package ctbrec.recorder.download.hls;
import static java.nio.file.StandardOpenOption.*;
import java.io.ByteArrayInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.time.Duration;
import java.time.Instant;
import java.time.ZonedDateTime;
import java.util.LinkedList;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.Arrays;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.taktik.mpegts.Streamer;
import org.taktik.mpegts.sinks.ByteChannelSink;
import org.taktik.mpegts.sinks.MTSSink;
import org.taktik.mpegts.sources.BlockingMultiMTSSource;
import org.taktik.mpegts.sources.InputStreamMTSSource;
import com.iheartradio.m3u8.ParseException;
import com.iheartradio.m3u8.PlaylistException;
import ctbrec.Config;
import ctbrec.Hmac;
import ctbrec.Model;
import ctbrec.MpegUtil;
import ctbrec.OS;
import ctbrec.io.HttpClient;
import ctbrec.io.HttpException;
import ctbrec.recorder.ProgressListener;
import okhttp3.Request;
import okhttp3.Response;
public class MergedHlsDownload extends AbstractHlsDownload {
import ctbrec.io.StreamRedirectThread;
public class MergedHlsDownload extends HlsDownload {
private static final Logger LOG = LoggerFactory.getLogger(MergedHlsDownload.class);
private static final boolean IGNORE_CACHE = true;
private BlockingMultiMTSSource multiSource;
private Thread mergeThread;
private Streamer streamer;
private ZonedDateTime splitRecStartTime;
private Config config;
private File targetFile;
private FileChannel fileChannel = null;
private boolean downloadFinished = false;
public MergedHlsDownload(HttpClient client) {
super(client);
}
@Override
public void init(Config config, Model model) {
this.config = config;
this.model = model;
targetFile = Config.getInstance().getFileForRecording(model, "ts");
}
public void postprocess(ctbrec.Recording recording) {
super.postprocess(recording);
File dir = new File(Config.getInstance().getSettings().recordingsDir, recording.getPath());
File playlist = new File(dir, "playlist.m3u8");
if (!playlist.exists()) {
super.generatePlaylist(recording);
}
@Override
public File getTarget() {
return targetFile;
}
public void start(String segmentPlaylistUri, File targetFile, ProgressListener progressListener) throws IOException {
try {
running = true;
super.startTime = Instant.now();
splitRecStartTime = ZonedDateTime.now();
mergeThread = createMergeThread(targetFile, progressListener, false);
LOG.debug("Merge thread started");
mergeThread.start();
if (Config.getInstance().getSettings().requireAuthentication) {
URL u = new URL(segmentPlaylistUri);
String path = u.getPath();
byte[] key = Config.getInstance().getSettings().key;
if (!Config.getInstance().getContextPath().isEmpty()) {
path = path.substring(Config.getInstance().getContextPath().length());
// @formatter:off
String[] cmdline = OS.getFFmpegCommand(
"-i", playlist.getAbsolutePath(),
"-c:v", "copy",
"-c:a", "copy",
"-movflags", "faststart",
"-f", "mp4",
new File(dir, "0merged.mp4").getAbsolutePath()
);
// @formatter:on
LOG.debug("Command line: {}", Arrays.toString(cmdline));
Process ffmpeg = Runtime.getRuntime().exec(cmdline);
new Thread(new StreamRedirectThread(ffmpeg.getInputStream(), System.out)).start(); // NOSONAR
new Thread(new StreamRedirectThread(ffmpeg.getErrorStream(), System.err)).start(); // NOSONAR
int exitCode = ffmpeg.waitFor();
if (exitCode == 0) {
recording.setPath(recording.getPath() + '/' + "0merged.mp4");
Files.delete(playlist.toPath());
File[] segments = dir.listFiles((directory, filename) -> filename.endsWith(".ts"));
for (File segment : segments) {
Files.delete(segment.toPath());
}
String hmac = Hmac.calculate(path, key);
segmentPlaylistUri = segmentPlaylistUri + "?hmac=" + hmac;
}
LOG.debug("Downloading segments");
downloadSegments(segmentPlaylistUri, false);
LOG.debug("Waiting for merge thread to finish");
mergeThread.join();
LOG.debug("Merge thread finished");
} catch (ParseException e) {
throw new IOException("Couldn't parse stream information", e);
} catch (PlaylistException e) {
throw new IOException("Couldn't parse HLS playlist", e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException("Couldn't wait for write thread to finish. Recording might be cut off", e);
} catch (InvalidKeyException | NoSuchAlgorithmException | IllegalStateException e) {
throw new IOException("Couldn't add HMAC to playlist url", e);
} finally {
try {
streamer.stop();
} catch (Exception e) {
LOG.error("Couldn't stop streamer", e);
}
downloadThreadPool.shutdown();
try {
LOG.debug("Waiting for last segments for {}", model);
downloadThreadPool.awaitTermination(60, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
downloadFinished = true;
LOG.debug("Download terminated for {}", segmentPlaylistUri);
}
}
@Override
public void start() throws IOException {
try {
if (!model.isOnline(IGNORE_CACHE)) {
throw new IOException(model.getName() + "'s room is not public");
}
running = true;
super.startTime = Instant.now();
splitRecStartTime = ZonedDateTime.now();
String segments = getSegmentPlaylistUrl(model);
mergeThread = createMergeThread(targetFile, null, true);
mergeThread.start();
if (segments != null) {
if (config.getSettings().splitRecordings > 0) {
LOG.debug("Splitting recordings every {} seconds", config.getSettings().splitRecordings);
}
downloadSegments(segments, true);
} else {
throw new IOException("Couldn't determine segments uri");
}
} catch (ParseException e) {
throw new IOException("Couldn't parse stream information", e);
} catch (PlaylistException e) {
throw new IOException("Couldn't parse HLS playlist", e);
} catch (EOFException e) {
// end of playlist reached
LOG.debug("Reached end of playlist for model {}", model);
} catch (Exception e) {
throw new IOException("Couldn't download segment", e);
} finally {
if (streamer != null) {
try {
streamer.stop();
} catch (Exception e) {
LOG.error("Couldn't stop streamer", e);
}
}
downloadThreadPool.shutdown();
try {
LOG.debug("Waiting for last segments for {}", model);
downloadThreadPool.awaitTermination(60, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
downloadFinished = true;
LOG.debug("Download for {} terminated", model);
}
}
private void downloadSegments(String segmentPlaylistUri, boolean livestreamDownload) throws IOException, ParseException, PlaylistException {
int lastSegment = 0;
int nextSegment = 0;
while (running) {
try {
SegmentPlaylist lsp = getNextSegments(segmentPlaylistUri);
emptyPlaylistCheck(lsp);
if (!livestreamDownload) {
multiSource.setTotalSegments(lsp.segments.size());
}
// download new segments
long downloadStart = System.currentTimeMillis();
if (livestreamDownload) {
downloadNewSegments(lsp, nextSegment);
} else {
downloadRecording(lsp);
}
long downloadTookMillis = System.currentTimeMillis() - downloadStart;
// download segments, which might have been skipped
if (nextSegment > 0 && lsp.seq > nextSegment) {
LOG.warn("Missed segments {} < {} in download for {}. Download took {}ms. Playlist is {}sec", nextSegment, lsp.seq, lsp.url,
downloadTookMillis, lsp.totalDuration);
}
if (livestreamDownload) {
// split up the recording, if configured
boolean split = splitRecording();
if (split) {
break;
}
// wait some time until requesting the segment playlist again to not hammer the server
waitForNewSegments(lsp, lastSegment, downloadTookMillis);
lastSegment = lsp.seq;
nextSegment = lastSegment + lsp.segments.size();
} else {
break;
}
} catch (HttpException e) {
if (e.getResponseCode() == 404) {
LOG.debug("Playlist not found (404). Model {} probably went offline", model);
} else if (e.getResponseCode() == 403) {
LOG.debug("Playlist access forbidden (403). Model {} probably went private or offline", model);
} else {
LOG.info("Unexpected error while downloading {}", model, e);
}
running = false;
} catch (Exception e) {
LOG.info("Unexpected error while downloading {}", model, e);
running = false;
}
}
}
private void downloadRecording(SegmentPlaylist lsp) throws IOException, InterruptedException {
for (String segment : lsp.segments) {
URL segmentUrl = new URL(segment);
SegmentDownload segmentDownload = new SegmentDownload(lsp, segmentUrl, client);
byte[] segmentData = segmentDownload.call();
writeSegment(segmentData);
}
}
private void downloadNewSegments(SegmentPlaylist lsp, int nextSegment) throws MalformedURLException, ExecutionException, HttpException {
int skip = nextSegment - lsp.seq;
// add segments to download threadpool
Queue<Future<byte[]>> downloads = new LinkedList<>();
if (downloadQueue.remainingCapacity() == 0) {
LOG.warn("Download to slow for this stream. Download queue is full. Skipping segment");
} else {
for (String segment : lsp.segments) {
if (!running) {
break;
}
if (skip > 0) {
skip--;
} else {
URL segmentUrl = new URL(segment);
Future<byte[]> download = downloadThreadPool.submit(new SegmentDownload(lsp, segmentUrl, client));
downloads.add(download);
}
}
}
// get completed downloads and write them to the file
// TODO it might be a good idea to do this in a separate thread, so that the main download loop isn't blocked
writeFinishedSegments(downloads);
}
private void writeFinishedSegments(Queue<Future<byte[]>> downloads) throws ExecutionException, HttpException {
for (Future<byte[]> downloadFuture : downloads) {
try {
byte[] segmentData = downloadFuture.get();
writeSegment(segmentData);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error("Error while downloading segment", e);
} catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause instanceof MissingSegmentException) {
if (model != null && !isModelOnline()) {
LOG.debug("Error while downloading segment, because model {} is offline. Stopping now", model.getName());
running = false;
} else {
LOG.debug("Segment not available, but model {} still online. Going on", Optional.ofNullable(model).map(Model::getName).orElse("n/a"));
}
} else if (cause instanceof HttpException) {
HttpException he = (HttpException) cause;
if (model != null && !isModelOnline()) {
LOG.debug("Error {} while downloading segment, because model {} is offline. Stopping now", he.getResponseCode(), model.getName());
running = false;
} else {
if (he.getResponseCode() == 404) {
LOG.info("Playlist for {} not found [HTTP 404]. Stopping now", Optional.ofNullable(model).map(Model::getName).orElse("n/a"));
running = false;
} else if (he.getResponseCode() == 403) {
LOG.info("Playlist for {} not accessible [HTTP 403]. Stopping now", Optional.ofNullable(model).map(Model::getName).orElse("n/a"));
running = false;
} else {
throw he;
}
}
} else {
throw e;
}
}
}
}
private void writeSegment(byte[] segmentData) throws InterruptedException {
InputStream in = new ByteArrayInputStream(segmentData);
InputStreamMTSSource source = InputStreamMTSSource.builder().setInputStream(in).build();
multiSource.addSource(source);
}
private boolean splitRecording() {
if (config.getSettings().splitRecordings > 0) {
Duration recordingDuration = Duration.between(splitRecStartTime, ZonedDateTime.now());
long seconds = recordingDuration.getSeconds();
if (seconds >= config.getSettings().splitRecordings) {
internalStop();
return true;
}
}
return false;
}
private void waitForNewSegments(SegmentPlaylist lsp, int lastSegment, long downloadTookMillis) {
try {
long wait = 0;
if (lastSegment == lsp.seq) {
int timeLeftMillis = (int) (lsp.totalDuration * 1000 - downloadTookMillis);
if (timeLeftMillis < 3000) { // we have less than 3 seconds to get the new playlist and start downloading it
wait = 1;
} else {
// wait a second to be nice to the server (don't hammer it with requests)
// 1 second seems to be a good compromise. every other calculation resulted in more missing segments
wait = 1000;
}
LOG.trace("Playlist didn't change... waiting for {}ms", wait);
} else {
// playlist did change -> wait for at least last segment duration
wait = 1;
LOG.trace("Playlist changed... waiting for {}ms", wait);
}
Thread.sleep(wait);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
if (running) {
LOG.error("Couldn't sleep between segment downloads. This might mess up the download!");
}
}
}
@Override
public void stop() {
if (running) {
try {
internalStop();
int count = 0;
while (!downloadFinished && count++ < 60) {
LOG.debug("Waiting for download to finish {}", model);
Thread.sleep(1000);
}
if(!downloadFinished) {
LOG.warn("Download didn't finish properly for model {}", model);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error("Couldn't wait for download to finish", e);
}
LOG.debug("Download stopped");
}
}
@Override
synchronized void internalStop() {
running = false;
if (streamer != null) {
streamer.stop();
streamer = null;
}
}
private Thread createMergeThread(File targetFile, ProgressListener listener, boolean liveStream) {
Thread t = new Thread(() -> {
multiSource = BlockingMultiMTSSource.builder().setFixContinuity(true).setProgressListener(listener).build();
try {
Path downloadDir = targetFile.getParentFile().toPath();
if (!downloadDir.toFile().exists()) {
Files.createDirectories(downloadDir);
}
fileChannel = FileChannel.open(targetFile.toPath(), CREATE, WRITE);
MTSSink sink = ByteChannelSink.builder().setByteChannel(fileChannel).build();
streamer = Streamer.builder().setSource(multiSource).setSink(sink).setSleepingEnabled(liveStream).setBufferSize(10)
.setName(Optional.ofNullable(model).map(Model::getName).orElse("")).build();
// Start streaming
streamer.stream();
LOG.debug("Streamer finished");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
if (running) {
LOG.error("Error while waiting for a download future", e);
}
} catch (Exception e) {
LOG.error("Error while saving stream to file", e);
} finally {
deleteEmptyRecording(targetFile);
running = false;
closeFile(fileChannel);
}
});
t.setName("Segment Merger Thread [" + model.getName() + "]");
t.setDaemon(true);
return t;
}
private void deleteEmptyRecording(File targetFile) {
try {
if (targetFile.exists() && targetFile.length() == 0) {
Files.delete(targetFile.toPath());
Files.delete(targetFile.getParentFile().toPath());
}
} catch (Exception e) {
LOG.error("Error while deleting empty recording {}", targetFile);
}
}
private void closeFile(FileChannel channel) {
try {
if (channel != null && channel.isOpen()) {
channel.close();
}
} catch (Exception e) {
LOG.error("Error while closing file channel", e);
}
}
private class SegmentDownload implements Callable<byte[]> {
private URL url;
private HttpClient client;
private SegmentPlaylist lsp;
public SegmentDownload(SegmentPlaylist lsp, URL url, HttpClient client) {
this.lsp = lsp;
this.url = url;
this.client = client;
}
@Override
public byte[] call() throws IOException {
LOG.trace("Downloading segment {}", url.getFile());
int maxTries = 3;
for (int i = 1; i <= maxTries && running; i++) {
Request request = new Request.Builder().url(url).addHeader("connection", "keep-alive").build();
try (Response response = client.execute(request)) {
if (response.isSuccessful()) {
byte[] segment = response.body().bytes();
if (lsp.encrypted) {
segment = new Crypto(lsp.encryptionKeyUrl, client).decrypt(segment);
}
return segment;
} else {
throw new HttpException(response.code(), response.message());
}
} catch (Exception e) {
if (i == maxTries) {
LOG.error("Error while downloading segment. Segment {} finally failed", url.getFile());
} else {
LOG.trace("Error while downloading segment {} on try {}", url.getFile(), i, e);
}
if (model != null && !isModelOnline()) {
break;
}
}
}
throw new MissingSegmentException("Unable to download segment " + url.getFile() + " after " + maxTries + " tries");
}
}
public boolean isModelOnline() {
try {
return model.isOnline(IGNORE_CACHE);
} catch (IOException | ExecutionException e) {
return false;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
@Override
public String getPath(Model model) {
String absolutePath = targetFile.getAbsolutePath();
String recordingsDir = Config.getInstance().getSettings().recordingsDir;
String relativePath = absolutePath.replaceFirst(Pattern.quote(recordingsDir), "");
return relativePath;
}
@Override
public Duration getLength() {
try {
return Duration.ofSeconds((long) MpegUtil.getFileDuration(targetFile));
LOG.error("Interrupted while waiting for FFMPEG", e);
} catch (IOException e) {
LOG.error("Couldn't determine recording length", e);
return Duration.ofSeconds(0);
LOG.error("Couldn't execute FFMPEG", e);
}
}
}