hl-valdemar/terebi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Releases Wiki Activity
Files
599c1492eeffdfe9509bbe5f5c3126698b1a5306
terebi/Sources/Core/cast_media.cpp
T

985 lines
56 KiB
C++
Raw Blame History

#include "cast_media.h"
#include "network_policy.hpp"
#include "network_input.hpp"
extern "C" {
#include <libavformat/avformat.h>
#include <libavcodec/avcodec.h>
#include <libavutil/audio_fifo.h>
#include <libavutil/opt.h>
#include <libswresample/swresample.h>
#include <libswscale/swscale.h>
}
#include <algorithm>
#include <array>
#include <atomic>
#include <charconv>
#include <chrono>
#include <cmath>
#include <condition_variable>
#include <cstdarg>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <deque>
#include <map>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <string>
#include <thread>
#include <vector>
#include <netdb.h>
#include <poll.h>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>
#include <limits>
namespace {
using Clock = std::chrono::steady_clock;
constexpr double segmentSeconds = 6;
constexpr size_t segmentLimit = 32 * 1024 * 1024;
constexpr size_t windowSegments = 8;
void check(int value) { if (value < 0) throw std::runtime_error("Cast media operation failed (private details redacted)."); }
void require(bool value) { if (!value) throw std::runtime_error("Cast media is unavailable or unsupported (private details redacted)."); }
void privateLog(void *, int level, const char *, va_list) {
if (level > AV_LOG_WARNING || level > av_log_get_level()) return;
std::fputs(level <= AV_LOG_ERROR ? "FFmpeg cast error (private details redacted)\n" :
"FFmpeg cast warning (private details redacted)\n", stderr);
}
struct FrameDelete { void operator()(AVFrame *p) const { av_frame_free(&p); } };
struct PacketDelete { void operator()(AVPacket *p) const { av_packet_free(&p); } };
using Frame = std::unique_ptr<AVFrame, FrameDelete>;
using Packet = std::unique_ptr<AVPacket, PacketDelete>;
Frame frame() { auto p = Frame(av_frame_alloc()); require(bool(p)); return p; }
Packet packet() { auto p = Packet(av_packet_alloc()); require(bool(p)); return p; }
int64_t aacBoundary(double seconds) { return int64_t(std::ceil(seconds * 48000 / 1024)) * 1024; }
double aacPreroll(double seconds) { return double(std::max<int64_t>(0, aacBoundary(seconds) - 12 * 1024)) / 48000; }
bool hasIDR(const AVPacket &packet, const AVCodecParameters &codec) {
const auto *bytes = packet.data;
const size_t size = size_t(std::max(0, packet.size));
if (codec.extradata_size >= 5 && codec.extradata[0] == 1) {
const size_t lengthSize = (codec.extradata[4] & 3) + 1;
for (size_t at = 0; at + lengthSize <= size;) {
uint32_t length = 0;
for (size_t i = 0; i < lengthSize; ++i) length = (length << 8) | bytes[at++];
if (!length || length > size - at) return false;
if ((bytes[at] & 31) == 5) return true;
at += length;
}
return false;
}
for (size_t at = 0; at + 3 < size; ++at) {
if (bytes[at] == 0 && bytes[at + 1] == 0 && bytes[at + 2] == 1 &&
(bytes[at + 3] & 31) == 5) return true;
}
return false;
}
struct Source {
std::string video, videoHeaders, audio, audioHeaders;
bool local, restricted, live;
double duration;
};
struct Input {
AVFormatContext *format = nullptr;
std::atomic<bool> &cancel;
std::string origin, headers;
std::atomic<int64_t> deadline{0};
std::atomic<int64_t> operationDeadline{std::numeric_limits<int64_t>::max()};
bool network, restricted, live;
std::atomic<uint64_t> epoch{0};
std::mutex waitLock;
std::condition_variable wake;
bool positioned = false;
std::unique_ptr<playback::NetworkInput> buffered;
Input(std::atomic<bool> &stop, const std::string &url, const std::string &fields, bool local, bool restricted, bool live) :
cancel(stop), headers(fields), network(!local), restricted(restricted), live(live) {
origin = local ? "" : playback::networkOrigin(url.c_str());
format = avformat_alloc_context(); require(format != nullptr);
format->interrupt_callback = {interrupt, this};
if (network) { format->opaque = this; format->io_open = openResource; }
AVDictionary *options = nullptr;
auto set = [&](const char *key, const char *value) {
int result = av_dict_set(&options, key, value, 0);
if (result < 0) { av_dict_free(&options); avformat_close_input(&format); check(result); }
};
set("protocol_whitelist", local ? "file" : "http,https,tcp,tls,crypto");
set("rw_timeout", live ? "15000000" : "5000000");
set("probesize", "5242880");
set("analyzeduration", "5000000");
if (network) {
if (!headers.empty()) set("headers", headers.c_str());
set("max_redirects", "0");
set("http_persistent", "0");
set("http_multiple", "0");
set("extension_picky", "0");
set("request_size", "10485760");
}
arm();
int result = avformat_open_input(&format, url.c_str(), nullptr, &options);
av_dict_free(&options);
if (result >= 0) { arm(); result = avformat_find_stream_info(format, nullptr); }
if (result < 0) { avformat_close_input(&format); check(result); }
}
~Input() {
if (buffered) buffered->stop();
else avformat_close_input(&format);
}
void arm() {
const auto ioDeadline = std::chrono::duration_cast<std::chrono::nanoseconds>((Clock::now() + std::chrono::seconds(15)).time_since_epoch()).count();
deadline = std::min<int64_t>(ioDeadline, operationDeadline.load());
}
bool interrupted() const {
const auto now = std::chrono::duration_cast<std::chrono::nanoseconds>(Clock::now().time_since_epoch()).count();
return cancel || now >= operationDeadline.load() || (!buffered && now >= deadline.load());
}
static int interrupt(void *opaque) {
auto &in = *static_cast<Input *>(opaque);
return in.interrupted() || (in.buffered && playback::NetworkInput::interrupt(in.buffered.get()));
}
static int openResource(AVFormatContext *context, AVIOContext **io, const char *url, int flags, AVDictionary **options) noexcept {
try {
auto &in = *static_cast<Input *>(context->opaque);
const char *resource = std::strncmp(url, "crypto+", 7) == 0 || std::strncmp(url, "crypto:", 7) == 0 ? url + 7 : url;
const auto resourceOrigin = playback::networkOrigin(resource);
if (resourceOrigin.empty() || (in.restricted && resourceOrigin != in.origin)) return AVERROR(EACCES);
int result = av_dict_set(options, "protocol_whitelist", "http,https,tcp,tls,crypto", 0);
// Public YouTube HLS may cross CDN origins; never carry private headers across them.
if (result >= 0) result = av_dict_set(options, "headers", resourceOrigin == in.origin && !in.headers.empty() ? in.headers.c_str() : nullptr, 0);
if (result >= 0) result = av_dict_set(options, "max_redirects", "0", 0);
if (result >= 0) result = av_dict_set(options, "rw_timeout", in.live ? "15000000" : "5000000", 0);
if (result < 0) return result;
return avio_open2(io, url, flags, &context->interrupt_callback, options);
} catch (...) { return AVERROR(EACCES); }
}
double offset() const { return format->start_time == AV_NOPTS_VALUE ? 0 : double(format->start_time) / AV_TIME_BASE; }
double duration() const { return format->duration == AV_NOPTS_VALUE ? 0 : double(format->duration) / AV_TIME_BASE; }
void select(std::array<int, 2> indices, playback::StreamCacheBudget &budget) {
for (unsigned i = 0; i < format->nb_streams; ++i)
format->streams[i]->discard = int(i) == indices[0] || int(i) == indices[1] ?
AVDISCARD_DEFAULT : AVDISCARD_ALL;
if (!network || live) return;
buffered = std::make_unique<playback::NetworkInput>(cancel, epoch, 0, wake, budget);
buffered->format = format;
buffered->start(indices, offset(), true);
}
void seek(double seconds, double audioTarget = NAN) {
// Round forward so a sub-microsecond truncation cannot select the
// previous GOP when a playlist boundary is an exact stream timestamp.
const auto timestamp = int64_t(std::ceil((seconds + offset()) * AV_TIME_BASE));
if (buffered) {
// Probing retains the opening packets. Do not reopen their HLS media
// resource just to establish the initial zero-position cursors.
if (!positioned && seconds == 0) { positioned = true; return; }
positioned = true;
// A nearby forward request can outrun the asynchronous fetch frontier.
// Let the existing read complete rather than interrupting and refetching it.
while (buffered->readAheadPending(seconds, audioTarget)) {
require(!interrupted());
std::unique_lock lock(waitLock);
wake.wait_for(lock, std::chrono::milliseconds(20));
}
buffered->seek(-1, timestamp, audioTarget);
} else { arm(); check(av_seek_frame(format, -1, timestamp, AVSEEK_FLAG_BACKWARD)); }
}
int read(int slot, AVPacket *destination) {
if (!buffered) return av_read_frame(format, destination);
for (;;) {
require(!interrupted());
double end = 0;
int result = buffered->take(slot, destination, end);
if (result != 0) return result == 1 ? 0 : result;
std::unique_lock lock(waitLock);
wake.wait_for(lock, std::chrono::milliseconds(20));
}
}
};
bool remuxAudio(const AVCodecParameters &codec) {
return codec.codec_id == AV_CODEC_ID_AAC && codec.ch_layout.nb_channels <= 2 &&
(codec.profile == AV_PROFILE_AAC_LOW || codec.profile == AV_PROFILE_AAC_HE || codec.profile == AV_PROFILE_AAC_HE_V2);
}
bool remuxVideo(Input &input, AVStream &stream) {
const auto &codec = *stream.codecpar;
const int profile = codec.profile & ~AV_PROFILE_H264_CONSTRAINED;
const auto rate = av_guess_frame_rate(input.format, &stream, nullptr);
return codec.codec_id == AV_CODEC_ID_H264 && codec.video_delay == 0 &&
(profile == AV_PROFILE_H264_BASELINE || profile == AV_PROFILE_H264_MAIN || profile == AV_PROFILE_H264_HIGH) &&
codec.level > 0 && codec.level <= 41 && codec.width > 0 && codec.width <= 1920 &&
codec.height > 0 && codec.height <= 1080 && rate.num > 0 && av_q2d(rate) <= 30.001;
}
struct Track {
Input *input;
AVStream *stream = nullptr;
AVCodecContext *decoder = nullptr;
Packet pending;
Frame decoded;
bool eof = false, draining = false, remux = false;
double nextTime = 0;
Track(Input &in, int index, AVMediaType type, double start, bool live, bool aligned = false) : input(&in) {
require(index >= 0);
stream = input->format->streams[index];
remux = type == AVMEDIA_TYPE_VIDEO ? remuxVideo(in, *stream) : remuxAudio(*stream->codecpar);
const double seekTarget = type == AVMEDIA_TYPE_AUDIO && !remux ?
std::max(0.0, aacPreroll(start) - double(std::max(0, stream->codecpar->seek_preroll)) /
std::max(1, stream->codecpar->sample_rate)) : start;
if (!live && !input->buffered) input->seek(seekTarget);
nextPacket();
// Indexed VOD can advertise source GOP boundaries; every actual start
// still has to be an IDR. The opening frame may follow AAC priming.
if (type == AVMEDIA_TYPE_VIDEO) {
// Cached seeks can return the preceding GOP while read-ahead has
// reached this boundary but not yet delivered its IDR packet.
if (aligned && remux)
while (pending && packetTime() < start - .001) nextPacket();
const auto rate = av_guess_frame_rate(input->format, stream, nullptr);
const double openingGap = rate.num > 0 ? av_q2d(av_inv_q(rate)) : 0;
remux = remux && !live && pending && (pending->flags & AV_PKT_FLAG_KEY) &&
hasIDR(*pending, *stream->codecpar) &&
(std::abs(packetTime() - start) < .001 ||
(start == 0 && packetTime() >= 0 && packetTime() <= openingGap));
if (aligned && !remux) {
// A keyframe index is not proof of an IDR; recover full decoder preroll.
input->seek(0); eof = false; nextPacket();
}
}
if (!live && type == AVMEDIA_TYPE_VIDEO && pending && (pending->flags & AV_PKT_FLAG_KEY) &&
!hasIDR(*pending, *stream->codecpar) && stream->codecpar->codec_id == AV_CODEC_ID_H264) {
// ponytail: rare intra-refresh seeks decode from the beginning; index real IDRs if this becomes slow.
input->seek(0);
nextPacket();
}
if (!remux) {
const auto *codec = avcodec_find_decoder(stream->codecpar->codec_id); require(codec != nullptr);
decoder = avcodec_alloc_context3(codec); require(decoder != nullptr);
try {
require(stream->codecpar->width >= 0 && stream->codecpar->height >= 0 &&
int64_t(stream->codecpar->width) * stream->codecpar->height <= 4096LL * 2160);
check(avcodec_parameters_to_context(decoder, stream->codecpar));
decoder->pkt_timebase = stream->time_base;
decoder->thread_count = 2;
decoder->max_pixels = 4096LL * 2160;
check(avcodec_open2(decoder, codec, nullptr));
} catch (...) { avcodec_free_context(&decoder); throw; }
}
}
~Track() { avcodec_free_context(&decoder); }
double packetTime() const {
if (!pending) return INFINITY;
auto pts = pending->pts != AV_NOPTS_VALUE ? pending->pts : pending->dts;
return pts == AV_NOPTS_VALUE ? nextTime : pts * av_q2d(stream->time_base) - input->offset();
}
void nextPacket() {
pending.reset();
if (eof || !stream) return;
auto p = packet();
input->arm();
for (;;) {
require(!input->interrupted());
int result = input->read(stream->codecpar->codec_type == AVMEDIA_TYPE_AUDIO ? 0 : 1, p.get());
if (result == AVERROR_EOF) { eof = true; return; }
check(result);
require(p->size >= 0 && size_t(p->size) <= segmentLimit);
if (p->stream_index == stream->index) { pending = std::move(p); return; }
av_packet_unref(p.get());
}
}
double frameTime() const {
if (!decoded) return INFINITY;
auto pts = decoded->best_effort_timestamp;
return pts == AV_NOPTS_VALUE ? nextTime : pts * av_q2d(stream->time_base) - input->offset();
}
void nextFrame() {
decoded.reset(); auto f = frame();
for (;;) {
require(!input->interrupted());
int result = avcodec_receive_frame(decoder, f.get());
if (result == 0) {
decoded = std::move(f);
const double step = stream->codecpar->codec_type == AVMEDIA_TYPE_AUDIO ? double(decoded->nb_samples) / decoded->sample_rate : 1.0 / 30;
nextTime = frameTime() + step;
return;
}
if (result == AVERROR_EOF) return;
if (result != AVERROR(EAGAIN)) check(result);
if (!pending) {
if (draining) return;
check(avcodec_send_packet(decoder, nullptr)); draining = true;
} else {
check(avcodec_send_packet(decoder, pending.get())); nextPacket();
}
}
}
double time() { if (remux) return packetTime(); if (!decoded) nextFrame(); return frameTime(); }
};
struct Segment { int64_t id; double duration; std::vector<uint8_t> bytes; };
struct Mux {
AVFormatContext *format = nullptr;
AVIOContext *io = nullptr;
std::vector<uint8_t> bytes;
Mux() {
check(avformat_alloc_output_context2(&format, nullptr, "mpegts", nullptr));
auto *buffer = static_cast<uint8_t *>(av_malloc(32768));
if (!buffer) { avformat_free_context(format); format = nullptr; throw std::bad_alloc(); }
io = avio_alloc_context(buffer, 32768, 1, this, nullptr, writeBytes, nullptr);
if (!io) { av_free(buffer); avformat_free_context(format); format = nullptr; throw std::bad_alloc(); }
format->pb = io; format->flags |= AVFMT_FLAG_CUSTOM_IO;
}
~Mux() { if (io) { av_freep(&io->buffer); avio_context_free(&io); } avformat_free_context(format); }
static int writeBytes(void *opaque, const uint8_t *data, int count) {
auto &self = *static_cast<Mux *>(opaque);
if (count < 0 || self.bytes.size() + size_t(count) > segmentLimit) return AVERROR(ENOSPC);
try { self.bytes.insert(self.bytes.end(), data, data + count); return count; } catch (...) { return AVERROR(ENOMEM); }
}
};
struct OutputTrack {
Track &source;
Mux *mux;
AVStream *stream = nullptr;
AVCodecContext *encoder = nullptr;
SwsContext *scale = nullptr;
SwrContext *resample = nullptr;
AVAudioFifo *fifo = nullptr;
int64_t audioPTS = AV_NOPTS_VALUE;
bool firstVideo = true;
double lastVideoTime = -INFINITY;
bool live, drained = false;
std::deque<Packet> packets;
size_t packetBytes = 0;
int64_t lastPacketPTS = AV_NOPTS_VALUE;
int64_t keepFrom = INT64_MIN, keepUntil = INT64_MAX;
OutputTrack(Track &track, Mux &output, bool live) : source(track), mux(&output), live(live) {
try {
if (!source.stream) return;
stream = avformat_new_stream(mux->format, nullptr); require(stream != nullptr);
if (source.remux) {
check(avcodec_parameters_copy(stream->codecpar, source.stream->codecpar));
stream->time_base = source.stream->time_base; stream->codecpar->codec_tag = 0;
return;
}
bool video = source.stream->codecpar->codec_type == AVMEDIA_TYPE_VIDEO;
const auto *codec = video ? avcodec_find_encoder_by_name("h264_videotoolbox") : avcodec_find_encoder(AV_CODEC_ID_AAC);
require(codec != nullptr); encoder = avcodec_alloc_context3(codec); require(encoder != nullptr);
if (video) {
require(source.decoder->width > 0 && source.decoder->height > 0 &&
int64_t(source.decoder->width) * source.decoder->height <= 4096LL * 2160);
auto sar = av_guess_sample_aspect_ratio(source.input->format, source.stream, nullptr);
if (sar.num <= 0 || sar.den <= 0) sar = {1, 1};
const double displayWidth = source.decoder->width * av_q2d(sar);
const double ratio = std::min({1.0, 1920.0 / displayWidth, 1080.0 / source.decoder->height});
encoder->width = int(displayWidth * ratio) & ~1;
encoder->height = int(source.decoder->height * ratio) & ~1;
require(encoder->width > 0 && encoder->height > 0);
// VideoToolbox can omit SAR in its H264 bitstream; encode square pixels instead.
encoder->sample_aspect_ratio = {1, 1};
encoder->pix_fmt = AV_PIX_FMT_NV12; encoder->time_base = {1, 90000};
encoder->framerate = av_guess_frame_rate(source.input->format, source.stream, nullptr);
if (encoder->framerate.num <= 0 || av_q2d(encoder->framerate) > 30) encoder->framerate = {30, 1};
encoder->bit_rate = 6000000; encoder->gop_size = 180; encoder->max_b_frames = 0;
check(av_opt_set(encoder->priv_data, "profile", "main", 0));
check(av_opt_set(encoder->priv_data, "level", "4.1", 0));
// VideoToolbox may use its platform software fallback, never a host executable.
av_opt_set(encoder->priv_data, "allow_sw", "1", 0);
av_opt_set(encoder->priv_data, "realtime", "1", 0);
} else {
encoder->sample_rate = 48000; encoder->sample_fmt = AV_SAMPLE_FMT_FLTP;
av_channel_layout_default(&encoder->ch_layout, 2); encoder->time_base = {1, 48000}; encoder->bit_rate = 192000;
fifo = av_audio_fifo_alloc(encoder->sample_fmt, 2, 4096); require(fifo != nullptr);
}
check(avcodec_open2(encoder, codec, nullptr));
stream->time_base = encoder->time_base; check(avcodec_parameters_from_context(stream->codecpar, encoder));
} catch (...) { avcodec_free_context(&encoder); av_audio_fifo_free(fifo); fifo = nullptr; throw; }
}
~OutputTrack() { avcodec_free_context(&encoder); sws_freeContext(scale); swr_free(&resample); av_audio_fifo_free(fifo); }
void beginSegment(Mux &output) {
mux = &output;
stream = avformat_new_stream(mux->format, nullptr); require(stream != nullptr);
if (source.remux) {
check(avcodec_parameters_copy(stream->codecpar, source.stream->codecpar));
stream->time_base = source.stream->time_base; stream->codecpar->codec_tag = 0;
} else {
stream->time_base = encoder->time_base;
check(avcodec_parameters_from_context(stream->codecpar, encoder));
}
}
void writePacket(AVPacket *packet) {
// A shared, positive TS clock origin prevents per-segment AAC priming shifts.
const auto offset = av_rescale_q(1, AVRational{1, 1}, stream->time_base);
if (packet->pts != AV_NOPTS_VALUE) packet->pts += offset;
if (packet->dts != AV_NOPTS_VALUE) packet->dts += offset;
packet->stream_index = stream->index;
check(av_interleaved_write_frame(mux->format, packet));
}
void encoded(AVFrame *f) {
check(avcodec_send_frame(encoder, f)); auto p = packet();
for (;;) {
int result = avcodec_receive_packet(encoder, p.get());
if (result == AVERROR(EAGAIN) || result == AVERROR_EOF) break;
check(result);
lastPacketPTS = p->pts;
if (!live) {
require(p->pts != AV_NOPTS_VALUE && p->size >= 0);
if (p->pts >= keepUntil) {
require(packetBytes + size_t(p->size) <= segmentLimit);
packetBytes += size_t(p->size);
packets.push_back(std::move(p)); p = packet();
continue;
}
if (p->pts < keepFrom) { av_packet_unref(p.get()); continue; }
}
av_packet_rescale_ts(p.get(), encoder->time_base, stream->time_base);
writePacket(p.get()); av_packet_unref(p.get());
}
}
void audioFrames(bool finish, double end = INFINITY) {
const int count = encoder->frame_size;
while ((av_audio_fifo_size(fifo) >= count || (finish && av_audio_fifo_size(fifo) > 0)) &&
(audioPTS == AV_NOPTS_VALUE || double(audioPTS) / 48000 < end)) {
auto f = frame(); f->format = encoder->sample_fmt; f->sample_rate = encoder->sample_rate;
check(av_channel_layout_copy(&f->ch_layout, &encoder->ch_layout)); f->nb_samples = count;
check(av_frame_get_buffer(f.get(), 0));
int available = std::min(count, av_audio_fifo_size(fifo));
require(av_audio_fifo_read(fifo, reinterpret_cast<void **>(f->data), available) == available);
if (available < count) av_samples_set_silence(f->data, available, count - available, 2, encoder->sample_fmt);
f->pts = audioPTS; audioPTS += count; encoded(f.get());
}
}
void consume(double start, double end) {
if (source.remux) {
double time = source.packetTime();
if (time >= start - .001 && time < end) {
auto *p = source.pending.get();
int64_t offset = int64_t(source.input->offset() / av_q2d(source.stream->time_base));
if (p->pts != AV_NOPTS_VALUE) p->pts -= offset;
if (p->dts != AV_NOPTS_VALUE) p->dts -= offset;
av_packet_rescale_ts(p, source.stream->time_base, stream->time_base); p->stream_index = stream->index;
writePacket(p);
}
source.nextPacket(); return;
}
auto *f = source.decoded.get(); double time = source.frameTime();
if (source.stream->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
if (time >= start - .001 && time < end && time - lastVideoTime >= 1.0 / 30 - .00001) {
require(f->width > 0 && f->height > 0 && int64_t(f->width) * f->height <= 4096LL * 2160);
auto out = frame(); out->format = encoder->pix_fmt; out->width = encoder->width; out->height = encoder->height;
out->sample_aspect_ratio = encoder->sample_aspect_ratio;
check(av_frame_get_buffer(out.get(), 32));
scale = sws_getCachedContext(scale, f->width, f->height, AVPixelFormat(f->format), out->width, out->height, encoder->pix_fmt, SWS_BILINEAR, nullptr, nullptr, nullptr);
require(scale != nullptr); check(sws_scale(scale, f->data, f->linesize, 0, f->height, out->data, out->linesize));
out->pts = int64_t(std::llround(std::max(start, time) * 90000));
out->pict_type = firstVideo ? AV_PICTURE_TYPE_I : AV_PICTURE_TYPE_NONE; firstVideo = false; encoded(out.get());
lastVideoTime = time;
}
} else {
require(f->sample_rate > 0 && f->nb_samples >= 0 && f->nb_samples <= 65536 && f->ch_layout.nb_channels <= 32);
const double begin = live ? start : aacPreroll(start);
const double limit = end;
if (time + double(f->nb_samples) / f->sample_rate > begin && time < limit) {
if (!resample) {
check(swr_alloc_set_opts2(&resample, &encoder->ch_layout, encoder->sample_fmt, 48000, &f->ch_layout, AVSampleFormat(f->format), f->sample_rate, 0, nullptr));
check(swr_init(resample));
}
int capacity = int(av_rescale_rnd(swr_get_delay(resample, f->sample_rate) + f->nb_samples, 48000, f->sample_rate, AV_ROUND_UP));
require(capacity > 0 && capacity <= 131072);
auto out = frame(); out->format = encoder->sample_fmt; out->sample_rate = 48000;
check(av_channel_layout_copy(&out->ch_layout, &encoder->ch_layout)); out->nb_samples = capacity; check(av_frame_get_buffer(out.get(), 0));
int count = swr_convert(resample, out->data, capacity, const_cast<const uint8_t **>(f->extended_data), f->nb_samples); check(count);
int skip = audioPTS != AV_NOPTS_VALUE ? 0 :
std::clamp(int(std::llround((begin - time) * 48000)), 0, count);
int take = count - skip;
if (audioPTS == AV_NOPTS_VALUE) audioPTS = int64_t(std::llround(std::max(begin, time) * 48000));
uint8_t *data[2] = {out->data[0] + skip * 4, out->data[1] + skip * 4};
require(av_audio_fifo_size(fifo) + take <= 131072);
require(av_audio_fifo_write(fifo, reinterpret_cast<void **>(data), take) == take); audioFrames(false, end);
}
}
source.nextFrame();
}
void beginPackets(double start, double end) {
if (!encoder || live) return;
keepFrom = fifo ? aacBoundary(start) : int64_t(std::llround(start * 90000));
keepUntil = fifo ? int64_t(std::llround(end * 48000)) : int64_t(std::llround(end * 90000));
while (!packets.empty() && packets.front()->pts < keepUntil) {
auto p = std::move(packets.front()); packets.pop_front();
packetBytes -= size_t(p->size);
if (p->pts < keepFrom) continue;
av_packet_rescale_ts(p.get(), encoder->time_base, stream->time_base);
writePacket(p.get());
}
}
void finish(double start, double end) {
if (!encoder) return;
if (live) {
if (fifo) audioFrames(false, end); else encoded(nullptr);
return;
}
// Neither encoder can resume after a drain. Feed just enough future
// input to receive this segment's delayed packets; retain the rest.
const int64_t target = fifo ? aacBoundary(end) - encoder->frame_size :
(std::isfinite(lastVideoTime) ? int64_t(std::llround(lastVideoTime * 90000)) : AV_NOPTS_VALUE);
if (!fifo) firstVideo = true;
while (!drained && target != AV_NOPTS_VALUE && (lastPacketPTS == AV_NOPTS_VALUE || lastPacketPTS < target)) {
if (fifo) audioFrames(false);
if (lastPacketPTS != AV_NOPTS_VALUE && lastPacketPTS >= target) break;
const double time = source.time();
// ponytail: cap lookahead at six seconds; drain and reset on the
// next segment rather than following a sparse stream arbitrarily far.
if (!std::isfinite(time) || time >= end + segmentSeconds) {
if (fifo) {
// Drain resampler delay before padding the final AAC frame.
if (resample) {
auto tail = frame(); tail->format = encoder->sample_fmt;
tail->nb_samples = 4096;
check(av_channel_layout_copy(&tail->ch_layout, &encoder->ch_layout));
check(av_frame_get_buffer(tail.get(), 0));
int count = swr_convert(resample, tail->data, tail->nb_samples, nullptr, 0); check(count);
require(av_audio_fifo_write(fifo, reinterpret_cast<void **>(tail->data), count) == count);
}
audioFrames(true);
}
encoded(nullptr); drained = true;
break;
}
consume(fifo ? start : end, INFINITY);
}
}
};
struct Pipeline {
playback::StreamCacheBudget cacheBudget;
std::unique_ptr<Input> videoInput, audioInput;
Input *audioSource = nullptr;
int videoIndex = -1, audioIndex = -1;
Clock::time_point deadline;
std::unique_ptr<Track> video, audio;
std::unique_ptr<OutputTrack> videoOutput, audioOutput;
bool live;
std::atomic<bool> &cancel;
double duration = 0;
std::vector<double> boundaries;
double nextStart = 0;
Pipeline(const Source &source, std::atomic<bool> &stop, double start) : live(source.live), cancel(stop) {
videoInput = std::make_unique<Input>(stop, source.video, source.videoHeaders, source.local, source.restricted, source.live);
duration = std::max(source.duration, videoInput->duration());
videoIndex = av_find_best_stream(videoInput->format, AVMEDIA_TYPE_VIDEO, -1, -1, nullptr, 0);
if (!source.audio.empty()) {
require(videoIndex >= 0);
audioInput = std::make_unique<Input>(stop, source.audio, source.audioHeaders, false, source.restricted, source.live);
audioSource = audioInput.get();
audioIndex = av_find_best_stream(audioSource->format, AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0);
} else {
audioIndex = av_find_best_stream(videoInput->format, AVMEDIA_TYPE_AUDIO, -1, videoIndex, nullptr, 0);
if (audioIndex >= 0) {
if (videoIndex >= 0 && (source.local || source.live)) {
audioInput = std::make_unique<Input>(stop, source.video, source.videoHeaders, source.local, source.restricted, source.live);
audioIndex = av_find_best_stream(audioInput->format, AVMEDIA_TYPE_AUDIO, -1, videoIndex, nullptr, 0);
audioSource = audioInput.get();
} else audioSource = videoInput.get();
}
}
require(videoIndex >= 0 || audioIndex >= 0);
// Read only a complete, already-loaded sample index, before read-ahead
// owns the demuxer. No media scan or upstream seeks to build a playlist.
if (!live && videoIndex >= 0) {
auto *stream = videoInput->format->streams[videoIndex];
const int count = avformat_index_get_entries_count(stream);
if (remuxVideo(*videoInput, *stream) && count > 0 && stream->nb_frames == count) {
boundaries.push_back(0);
double previous = -INFINITY;
for (int i = 0; i < count; ++i) {
const auto *entry = avformat_index_get_entry(stream, i);
const double time = entry->timestamp * av_q2d(stream->time_base) - videoInput->offset();
if (entry->timestamp == AV_NOPTS_VALUE || time < previous) { boundaries.clear(); break; }
previous = time;
if (!(entry->flags & AVINDEX_KEYFRAME) || time < boundaries.back() + segmentSeconds) continue;
// ponytail: cap indexed GOP segments at twelve seconds; longer GOPs use the fixed-grid transcode.
if (time - boundaries.back() > 2 * segmentSeconds || time >= duration) break;
boundaries.push_back(time);
}
if (boundaries.size() < 2 || duration - boundaries.back() > 2 * segmentSeconds ||
duration <= boundaries.back()) boundaries.clear();
else boundaries.push_back(duration);
}
}
videoInput->select({audioSource == videoInput.get() ? audioIndex : -1, videoIndex}, cacheBudget);
if (audioInput) audioInput->select({audioIndex, -1}, cacheBudget);
reset(start, !boundaries.empty());
if (!source.audio.empty()) require(audio && audio->stream);
require(video || audio);
if (!source.live) require(std::isfinite(duration) && duration > 0 && duration <= 86400 * 14);
releaseOperation();
}
void armOperation() {
deadline = Clock::now() + std::chrono::seconds(45);
const auto limit = std::chrono::duration_cast<std::chrono::nanoseconds>(deadline.time_since_epoch()).count();
videoInput->operationDeadline = limit; videoInput->arm();
if (audioInput) { audioInput->operationDeadline = limit; audioInput->arm(); }
}
void releaseOperation() {
videoInput->operationDeadline = std::numeric_limits<int64_t>::max();
if (audioInput) audioInput->operationDeadline = std::numeric_limits<int64_t>::max();
}
void reset(double start, bool aligned = false) {
armOperation();
videoOutput.reset(); audioOutput.reset();
video.reset(); audio.reset();
// Random access resets codec history; the network demuxer rewinds both
// packet lanes atomically, seeking upstream only on cache misses.
const double audioTarget = audioIndex >= 0 && !remuxAudio(*audioSource->format->streams[audioIndex]->codecpar) ?
aacPreroll(start) : NAN;
if (videoInput->buffered)
videoInput->seek(start, audioSource == videoInput.get() ? audioTarget : NAN);
if (audioInput && audioInput->buffered) audioInput->seek(start, audioTarget);
if (videoIndex >= 0) video = std::make_unique<Track>(*videoInput, videoIndex, AVMEDIA_TYPE_VIDEO, start, live, aligned);
if (audioIndex >= 0) audio = std::make_unique<Track>(*audioSource, audioIndex, AVMEDIA_TYPE_AUDIO, start, live);
nextStart = start;
}
Segment segment(int64_t id, double start, double end) {
armOperation();
Mux mux;
if (video) {
if (live) videoOutput.reset();
if (videoOutput) videoOutput->beginSegment(mux);
else videoOutput = std::make_unique<OutputTrack>(*video, mux, live);
}
if (audio) {
if (audioOutput) audioOutput->beginSegment(mux);
else audioOutput = std::make_unique<OutputTrack>(*audio, mux, live);
}
AVDictionary *options = nullptr;
av_dict_set(&options, "mpegts_flags", "resend_headers", 0);
av_dict_set(&options, "mpegts_copyts", "1", 0);
av_dict_set(&options, "muxdelay", "0", 0);
int result = avformat_write_header(mux.format, &options); av_dict_free(&options); check(result);
if (videoOutput) videoOutput->beginPackets(start, end);
if (audioOutput) audioOutput->beginPackets(start, end);
const double audioEnd = audio && !audio->remux && !live ? double(aacBoundary(end)) / 48000 : end;
for (;;) {
require(!cancel && Clock::now() < deadline);
double vt = video ? video->time() : INFINITY, at = audio ? audio->time() : INFINITY;
if (vt >= end) vt = INFINITY;
if (at >= audioEnd) at = INFINITY;
if (!std::isfinite(vt) && !std::isfinite(at)) break;
if (vt <= at) videoOutput->consume(start, end); else audioOutput->consume(start, end);
}
if (videoOutput) videoOutput->finish(start, end);
if (audioOutput) audioOutput->finish(start, end);
check(av_write_trailer(mux.format)); avio_flush(mux.io);
require(mux.io->error >= 0 && !mux.bytes.empty());
if (videoInput->buffered) videoInput->buffered->updatePosition(end);
if (audioInput && audioInput->buffered) audioInput->buffered->updatePosition(end);
nextStart = end;
releaseOperation();
return {id, end - start, std::move(mux.bytes)};
}
};
bool number(std::string_view text, int64_t &value) {
if (text.empty()) return false;
auto result = std::from_chars(text.data(), text.data() + text.size(), value);
return result.ec == std::errc() && result.ptr == text.data() + text.size() && value >= 0;
}
}
struct CastMedia {
Source source;
std::atomic<bool> cancelled{false};
std::mutex lock;
std::mutex vodLock;
std::unique_ptr<Pipeline> vodPipeline;
std::condition_variable wake;
int listener = -1;
std::vector<int> clients;
std::array<std::thread, 2> servers;
std::thread producer;
std::string base, token;
double duration = 0;
std::vector<double> boundaries;
std::deque<std::shared_ptr<const Segment>> segments;
std::map<int, std::shared_ptr<const std::vector<uint8_t>>> subtitles;
#ifdef CAST_MEDIA_CHECK
double testSeconds = 6;
#endif
bool liveFailed = false;
explicit CastMedia(Source input) : source(std::move(input)) {}
~CastMedia() {
cancel();
if (producer.joinable()) producer.join();
for (auto &worker : servers) if (worker.joinable()) worker.join();
if (listener >= 0) close(listener);
}
void cancel() {
cancelled = true; wake.notify_all();
std::lock_guard guard(lock);
if (listener >= 0) shutdown(listener, SHUT_RDWR);
for (int socket : clients) shutdown(socket, SHUT_RDWR);
}
void bindReceiver(const char *host) {
require(host && std::strlen(host) <= 1024);
addrinfo hints{}; hints.ai_socktype = SOCK_DGRAM; hints.ai_family = AF_INET;
addrinfo *addresses = nullptr;
if (getaddrinfo(host, "8009", &hints, &addresses) != 0) {
hints.ai_family = AF_UNSPEC;
require(getaddrinfo(host, "8009", &hints, &addresses) == 0);
}
std::unique_ptr<addrinfo, decltype(&freeaddrinfo)> owner(addresses, freeaddrinfo);
sockaddr_storage local{}; socklen_t length = 0; int family = 0;
for (auto *address = addresses; address; address = address->ai_next) {
int route = socket(address->ai_family, SOCK_DGRAM, 0);
if (route < 0) continue;
length = sizeof(local);
bool found = connect(route, address->ai_addr, address->ai_addrlen) == 0 && getsockname(route, reinterpret_cast<sockaddr *>(&local), &length) == 0;
close(route);
if (found) { family = address->ai_family; break; }
}
require(family != 0 && !cancelled);
char name[NI_MAXHOST]; require(getnameinfo(reinterpret_cast<sockaddr *>(&local), length, name, sizeof(name), nullptr, 0, NI_NUMERICHOST) == 0);
if (family == AF_INET) reinterpret_cast<sockaddr_in *>(&local)->sin_port = 0;
else reinterpret_cast<sockaddr_in6 *>(&local)->sin6_port = 0;
int server = socket(family, SOCK_STREAM, 0); require(server >= 0);
if (fcntl(server, F_SETFL, O_NONBLOCK) < 0) { close(server); require(false); }
int noSIGPIPE = 1; setsockopt(server, SOL_SOCKET, SO_NOSIGPIPE, &noSIGPIPE, sizeof(noSIGPIPE));
if (::bind(server, reinterpret_cast<sockaddr *>(&local), length) != 0 || listen(server, 8) != 0 || getsockname(server, reinterpret_cast<sockaddr *>(&local), &length) != 0) { close(server); require(false); }
unsigned port = ntohs(family == AF_INET ? reinterpret_cast<sockaddr_in *>(&local)->sin_port : reinterpret_cast<sockaddr_in6 *>(&local)->sin6_port);
std::array<unsigned char, 32> random{}; arc4random_buf(random.data(), random.size());
for (auto byte : random) { char hex[3]; std::snprintf(hex, sizeof(hex), "%02x", byte); token += hex; }
std::string hostname = name;
// URLComponents expects escaped IPv6 scope identifiers.
auto scope = hostname.find('%'); if (scope != std::string::npos) hostname.replace(scope, 1, "%25");
base = "http://" + (family == AF_INET6 ? "[" + hostname + "]" : hostname) + ":" + std::to_string(port) + "/" + token + "/";
std::lock_guard guard(lock); listener = server;
if (cancelled) shutdown(listener, SHUT_RDWR);
}
void produce(std::unique_ptr<Pipeline> pipeline) {
try {
double first = std::min(pipeline->video ? pipeline->video->time() : INFINITY, pipeline->audio ? pipeline->audio->time() : INFINITY);
require(std::isfinite(first));
#ifndef CAST_MEDIA_CHECK
auto origin = Clock::now();
#endif
for (int64_t id = 0; !cancelled; ++id) {
auto segment = pipeline->segment(id, first + id * segmentSeconds, first + (id + 1) * segmentSeconds);
{ std::lock_guard guard(lock); segments.push_back(std::make_shared<const Segment>(std::move(segment))); while (segments.size() > windowSegments) segments.pop_front(); }
wake.notify_all();
std::unique_lock guard(lock);
#ifdef CAST_MEDIA_CHECK
wake.wait(guard, [&] { return cancelled || testSeconds >= (id + 1) * segmentSeconds; });
#else
wake.wait_until(guard, origin + std::chrono::milliseconds(int64_t((id + 1) * segmentSeconds * 1000)), [&] { return bool(cancelled); });
#endif
}
} catch (...) { std::lock_guard guard(lock); liveFailed = true; wake.notify_all(); }
}
int64_t segmentCount() const {
return boundaries.empty() ? int64_t(std::ceil(duration / segmentSeconds)) : int64_t(boundaries.size() - 1);
}
double boundary(int64_t id) const {
return boundaries.empty() ? std::min(duration, id * segmentSeconds) : boundaries[size_t(id)];
}
std::vector<uint8_t> playlist() {
double longest = segmentSeconds;
for (size_t i = 1; i < boundaries.size(); ++i)
longest = std::max(longest, boundaries[i] - boundaries[i - 1]);
std::string text = "#EXTM3U\n#EXT-X-VERSION:3\n#EXT-X-TARGETDURATION:" +
std::to_string(int(std::ceil(longest))) + "\n";
int64_t first = 0, count = segmentCount();
std::lock_guard guard(lock);
if (source.live) { require(!segments.empty() && !liveFailed); first = segments.front()->id; count = int64_t(segments.size()); }
else text += "#EXT-X-PLAYLIST-TYPE:VOD\n";
text += "#EXT-X-MEDIA-SEQUENCE:" + std::to_string(first) + "\n";
if (source.live) text += "#EXT-X-DISCONTINUITY-SEQUENCE:" + std::to_string(first) + "\n";
for (int64_t id = first; id < first + count; ++id) {
char value[64]; double length = source.live ? segments[size_t(id - first)]->duration : boundary(id + 1) - boundary(id);
std::snprintf(value, sizeof(value), "#EXTINF:%.6f,\n", length);
// Each segment retains an independent mux and random-access entry.
if (id > first) text += "#EXT-X-DISCONTINUITY\n";
text += value; text += "segment/" + std::to_string(id) + ".ts\n";
}
if (!source.live) text += "#EXT-X-ENDLIST\n";
return {text.begin(), text.end()};
}
std::shared_ptr<const Segment> vodSegment(int64_t id) {
auto cached = [&]() -> std::shared_ptr<const Segment> {
std::lock_guard guard(lock);
auto found = std::find_if(segments.begin(), segments.end(), [&](const auto &segment) { return segment->id == id; });
return found == segments.end() ? nullptr : *found;
};
if (auto segment = cached()) return segment;
// ponytail: uncached VOD generation is serialized; use a bounded input
// pool only if parallel uncached requests justify duplicate probe state.
std::lock_guard generation(vodLock);
require(!cancelled);
if (auto segment = cached()) return segment;
try {
const double start = boundary(id), end = boundary(id + 1);
if (vodPipeline) {
const auto &video = vodPipeline->video;
// Fixed-grid remux is safe only when the next boundary is an
// actual IDR. Otherwise reset once into the transcode path.
const bool independent = !video || !video->remux ||
(video->pending && std::abs(video->packetTime() - start) < .001 &&
(video->pending->flags & AV_PKT_FLAG_KEY) &&
hasIDR(*video->pending, *video->stream->codecpar));
const bool drained = (vodPipeline->videoOutput && vodPipeline->videoOutput->drained) ||
(vodPipeline->audioOutput && vodPipeline->audioOutput->drained);
if (vodPipeline->nextStart != start || !independent || drained)
vodPipeline->reset(start, !boundaries.empty());
} else vodPipeline = std::make_unique<Pipeline>(source, cancelled, start);
auto segment = std::make_shared<const Segment>(vodPipeline->segment(id, start, end));
{ std::lock_guard guard(lock); segments.push_back(segment); while (segments.size() > windowSegments) segments.pop_front(); }
return segment;
} catch (...) {
// A failed demux/decode is never cached or reused. A later receiver
// request can prepare fresh inputs; this request still receives 503.
vodPipeline.reset();
throw;
}
}
void sendAll(int socket, const uint8_t *bytes, size_t size) {
const auto deadline = Clock::now() + std::chrono::seconds(15);
while (size) {
require(!cancelled && Clock::now() < deadline);
pollfd descriptor{socket, POLLOUT, 0}; require(poll(&descriptor, 1, 1000) >= 0);
if (!(descriptor.revents & POLLOUT)) continue;
ssize_t count = send(socket, bytes, std::min(size, size_t(65536)), 0); require(count > 0);
bytes += count; size -= size_t(count);
}
}
void response(int socket, int status, const char *type, const std::vector<uint8_t> &bytes, bool head, std::string_view range = {}) {
size_t begin = 0, end = bytes.size(); std::string extra;
if (!range.empty() && status == 200) {
int64_t low = 0, high = 0; auto split = range.find('-');
if (!range.starts_with("bytes=") || split == std::string_view::npos || range.find(',') != std::string_view::npos) status = 416;
else if (split == 6) {
if (!number(range.substr(7), high) || high == 0 || bytes.empty()) status = 416;
else begin = bytes.size() - std::min(bytes.size(), size_t(high));
} else if (!number(range.substr(6, split - 6), low) || size_t(low) >= bytes.size()) status = 416;
else {
begin = size_t(low);
if (split + 1 < range.size()) {
if (!number(range.substr(split + 1), high) || high < low) status = 416;
else end = std::min(bytes.size(), size_t(high) + 1);
}
}
if (status != 416) { status = 206; extra = "Content-Range: bytes " + std::to_string(begin) + "-" + std::to_string(end - 1) + "/" + std::to_string(bytes.size()) + "\r\n"; }
else { extra = "Content-Range: bytes */" + std::to_string(bytes.size()) + "\r\n"; begin = end = 0; }
}
std::string header = "HTTP/1.1 " + std::to_string(status) + (status == 200 ? " OK" : status == 206 ? " Partial Content" : " Error") + "\r\nConnection: close\r\nAccess-Control-Allow-Origin: *\r\nAccess-Control-Allow-Methods: GET, HEAD, OPTIONS\r\nAccess-Control-Allow-Headers: Range\r\nAccess-Control-Expose-Headers: Content-Length, Content-Range, Accept-Ranges\r\nCache-Control: no-store\r\nAccept-Ranges: bytes\r\nContent-Type: " + type + "\r\n" + extra + "Content-Length: " + std::to_string(end - begin) + "\r\n\r\n";
sendAll(socket, reinterpret_cast<const uint8_t *>(header.data()), header.size());
if (!head && end > begin) sendAll(socket, bytes.data() + begin, end - begin);
}
void serve(int socket) {
std::string request; std::array<char, 2048> buffer{};
auto deadline = Clock::now() + std::chrono::seconds(5);
while (request.find("\r\n\r\n") == std::string::npos) {
require(!cancelled && request.size() < 16384 && Clock::now() < deadline);
pollfd descriptor{socket, POLLIN, 0}; require(poll(&descriptor, 1, 500) >= 0);
if (!(descriptor.revents & POLLIN)) continue;
ssize_t size = recv(socket, buffer.data(), std::min(buffer.size(), 16384 - request.size()), 0); require(size > 0); request.append(buffer.data(), size_t(size));
}
auto first = request.find(' '), second = request.find(' ', first == std::string::npos ? 0 : first + 1);
require(first != std::string::npos && second != std::string::npos && second < request.find("\r\n"));
auto method = request.substr(0, first), path = request.substr(first + 1, second - first - 1);
bool head = method == "HEAD";
if (!path.starts_with("/" + token + "/")) { response(socket, 404, "text/plain", {}, head); return; }
if (method == "OPTIONS") { response(socket, 204, "text/plain", {}, true); return; }
if (method != "GET" && !head) { response(socket, 405, "text/plain", {}, true); return; }
std::string range;
size_t cursor = request.find("\r\n") + 2;
while (cursor < request.size()) {
auto next = request.find("\r\n", cursor); if (next == cursor || next == std::string::npos) break;
auto line = request.substr(cursor, next - cursor); auto colon = line.find(':'); require(colon != std::string::npos);
auto name = line.substr(0, colon); for (char &c : name) if (c >= 'A' && c <= 'Z') c += 'a' - 'A';
if (name == "range") { require(range.empty()); auto begin = line.find_first_not_of(' ', colon + 1); require(begin != std::string::npos); range = line.substr(begin); }
cursor = next + 2;
}
path.erase(0, token.size() + 2);
std::vector<uint8_t> data; const char *type = "video/mp2t"; int status = 200;
std::shared_ptr<const std::vector<uint8_t>> retained;
if (path == "index.m3u8") { data = playlist(); type = "application/vnd.apple.mpegurl"; }
else if (path.starts_with("subtitle/") && path.ends_with(".vtt")) {
int64_t id = 0;
if (!number(std::string_view(path).substr(9, path.size() - 13), id) || id > 32) status = 404;
else { std::lock_guard guard(lock); auto found = subtitles.find(int(id)); if (found == subtitles.end()) status = 404; else retained = found->second; }
type = "text/vtt; charset=utf-8";
} else if (path.starts_with("segment/") && path.ends_with(".ts")) {
int64_t id = 0;
if (!number(std::string_view(path).substr(8, path.size() - 11), id)) status = 404;
else if (source.live) {
std::lock_guard guard(lock); auto found = std::find_if(segments.begin(), segments.end(), [&](const auto &segment) { return segment->id == id; });
if (found == segments.end()) status = 404; else retained = std::shared_ptr<const std::vector<uint8_t>>(*found, &(*found)->bytes);
} else if (id >= segmentCount()) status = 404;
else {
auto segment = vodSegment(id);
retained = std::shared_ptr<const std::vector<uint8_t>>(segment, &segment->bytes);
}
} else status = 404;
response(socket, status, type, retained ? *retained : data, head, range);
}
void server() {
while (!cancelled) {
pollfd descriptor{listener, POLLIN, 0}; if (poll(&descriptor, 1, 500) <= 0) continue;
if (!(descriptor.revents & POLLIN)) continue;
int socket = accept(listener, nullptr, nullptr); if (socket < 0) continue;
int one = 1; setsockopt(socket, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one));
timeval timeout{1, 0}; setsockopt(socket, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout)); setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout));
{ std::lock_guard guard(lock); clients.push_back(socket); if (cancelled) shutdown(socket, SHUT_RDWR); }
try { serve(socket); } catch (...) { try { response(socket, 503, "text/plain", {}, true); } catch (...) {} }
{ std::lock_guard guard(lock); clients.erase(std::remove(clients.begin(), clients.end(), socket), clients.end()); close(socket); }
}
}
};
extern "C" CastMedia *cast_media_create(const CastMediaSource *input, char error[512]) {
try {
require(input && input->video_url && std::strlen(input->video_url) <= 65536);
require(playback::validHeaders(input->video_headers) && playback::validHeaders(input->audio_headers));
require(std::isfinite(input->duration) && input->duration >= 0);
std::string video = input->video_url;
if (input->local) require(!video.empty() && video.front() == '/' && video.find('\0') == std::string::npos);
else require(playback::validNetworkURL(video.c_str()));
if (input->audio_url && *input->audio_url) require(playback::validNetworkURL(input->audio_url));
av_log_set_callback(privateLog);
// Bound demuxer-controlled allocations before probing or packet reads allocate them.
av_max_alloc(128 * 1024 * 1024);
return new CastMedia({video, input->video_headers ? input->video_headers : "", input->audio_url ? input->audio_url : "", input->audio_headers ? input->audio_headers : "", input->local, input->restricted, input->live, input->duration});
} catch (...) { if (error) std::snprintf(error, 512, "Could not prepare cast media (private details redacted)."); return nullptr; }
}
extern "C" bool cast_media_start(CastMedia *media, const char *host, char url[512], double *duration, char error[512]) {
try {
require(media && host && url && duration && !media->cancelled);
auto pipeline = std::make_unique<Pipeline>(media->source, media->cancelled, 0);
media->duration = pipeline->duration;
media->boundaries = std::move(pipeline->boundaries);
media->bindReceiver(host);
if (media->source.live) {
media->producer = std::thread([media, pipe = std::move(pipeline)]() mutable { media->produce(std::move(pipe)); });
std::unique_lock guard(media->lock);
require(media->wake.wait_for(guard, std::chrono::seconds(50), [&] { return media->segments.size() >= 2 || media->liveFailed || media->cancelled; }));
require(media->segments.size() >= 2 && !media->liveFailed && !media->cancelled);
} else media->vodPipeline = std::move(pipeline);
for (auto &worker : media->servers) worker = std::thread([media] { media->server(); });
require(media->base.size() + 11 < 512); std::snprintf(url, 512, "%sindex.m3u8", media->base.c_str()); *duration = media->source.live ? 0 : media->duration;
return true;
} catch (...) { if (media) media->cancel(); if (error) std::snprintf(error, 512, "Could not start cast delivery (private details redacted)."); return false; }
}
extern "C" bool cast_media_subtitle(CastMedia *media, int id, const void *bytes, size_t size, char url[512]) {
try {
require(media && !media->cancelled && id > 0 && id <= 32 && bytes && size > 0 && size <= 4 * 1024 * 1024 && url);
std::lock_guard guard(media->lock);
auto *data = static_cast<const uint8_t *>(bytes); media->subtitles[id] = std::make_shared<const std::vector<uint8_t>>(data, data + size);
require(media->base.size() + 32 < 512); std::snprintf(url, 512, "%ssubtitle/%d.vtt", media->base.c_str(), id); return true;
} catch (...) { return false; }
}
extern "C" void cast_media_cancel(CastMedia *media) { if (media) media->cancel(); }
extern "C" void cast_media_destroy(CastMedia *media) { delete media; }
#ifdef CAST_MEDIA_CHECK
extern "C" bool cast_media_check_advance(CastMedia *media, double seconds) {
if (!media || !media->source.live || !std::isfinite(seconds) || seconds < 6 || seconds > 3600) return false;
std::unique_lock guard(media->lock);
media->testSeconds = seconds;
media->wake.notify_all();
const auto target = int64_t(seconds / segmentSeconds);
media->wake.wait(guard, [&] { return media->cancelled || media->liveFailed ||
(!media->segments.empty() && media->segments.back()->id >= target); });
return !media->cancelled && !media->liveFailed;
}
#endif
Reference in New Issue View Git Blame Copy Permalink