ctbrec/common/src/main/java/ctbrec/sites/showup/ShowupMergedDownload.java

package ctbrec.sites.showup;

import static ctbrec.io.HttpConstants.*;

import java.io.IOException;
import java.io.InputStream;
import java.time.Instant;
import java.util.concurrent.ExecutorService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.iheartradio.m3u8.ParseException;
import com.iheartradio.m3u8.PlaylistException;

import ctbrec.Config;
import ctbrec.Model;
import ctbrec.io.BandwidthMeter;
import ctbrec.io.HttpClient;
import ctbrec.io.HttpException;
import ctbrec.recorder.download.ProcessExitedUncleanException;
import ctbrec.recorder.download.hls.AbstractHlsDownload;
import ctbrec.recorder.download.hls.MergedFfmpegHlsDownload;
import ctbrec.recorder.download.hls.SegmentPlaylist;
import okhttp3.Request;
import okhttp3.Response;

public class ShowupMergedDownload extends MergedFfmpegHlsDownload {

    private static final transient Logger LOG = LoggerFactory.getLogger(ShowupMergedDownload.class);

    private transient Response response;
    private transient InputStream in;
    private transient byte[] buffer = new byte[10240];

    public ShowupMergedDownload(HttpClient client) {
        super(client);
    }

    @Override
    public void init(Config config, Model model, Instant startTime, ExecutorService executorService) throws IOException {
        super.init(config, model, startTime, executorService);
        try {
            if (segmentPlaylistUrl == null) {
                segmentPlaylistUrl = getSegmentPlaylistUrl(model);
            }
            startDownload(segmentPlaylistUrl);
        } catch (Exception e) {
            LOG.error("Error starting the stream", e);
            stop();
        }
    }

    protected void startDownload(String segmentPlaylistUri) throws IOException, ParseException, PlaylistException {
        try {
            SegmentPlaylist lsp = getNextSegments(segmentPlaylistUri);
            emptyPlaylistCheck(lsp);

            String segment = lsp.segments.get(0);
            Request request = new Request.Builder().url(segment)
                    .header(USER_AGENT, Config.getInstance().getSettings().httpUserAgent)
                    .header(CONNECTION, KEEP_ALIVE)
                    .build();
            response = client.execute(request);
            if (response.isSuccessful()) {
                in = response.body().byteStream();
            } else {
                throw new HttpException(response.code(), response.message());
            }
        } catch (HttpException e) {
            handleHttpException(e);
        } catch (Exception e) {
            LOG.info("Unexpected error while downloading {}", model, e);
            stop();
        }
    }

    @Override
    public AbstractHlsDownload call() throws Exception {
        if (!running) {
            return this;
        }

        try {
            if (!ffmpegProcess.isAlive()) {
                running = false;
                int exitValue = ffmpegProcess.exitValue();
                ffmpeg.shutdown(exitValue);
            }
        } catch (ProcessExitedUncleanException e) {
            LOG.error("FFmpeg exited unclean", e);
        }

        streamSegmentDataToFfmpeg();
        rescheduleTime = Instant.now().plusSeconds(1);
        return this;
    }

    @Override
    protected void streamSegmentDataToFfmpeg() {
        downloadExecutor.submit(() -> {
            if (!ffmpegStreamLock.tryLock()) {
                return;
            }
            try {
                int length = -1;
                boolean keepGoing = true;
                while ((length = in.read(buffer)) > 0 && keepGoing) {
                    BandwidthMeter.add(length);
                    ffmpegStdIn.write(buffer, 0, length);
                    keepGoing = running && !Thread.interrupted() && model.isOnline(true);
                    splittingStrategy.splitNecessary(this);
                }
                if (length == -1) {
                    LOG.info("End of stream reached for {}", model);
                    stop();
                }
                LOG.debug("loop finished");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                LOG.error("Error while streaming data to FFmpeg", e);
            } finally {
                ffmpegStreamLock.unlock();
            }
        });
    }

    @Override
    public void finalizeDownload() {
        super.finalizeDownload();
        try {
            in.close();
            response.close();
        } catch (IOException e) {
            LOG.error("Error while finalizing download", e);
        }
    }
}