#include "cast_media.h" #include "network_policy.hpp" #include "network_input.hpp" extern "C" { #include #include #include #include #include #include } #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include 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; using Packet = std::unique_ptr; 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(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 &cancel; std::string origin, headers; std::atomic deadline{0}; std::atomic operationDeadline{std::numeric_limits::max()}; bool network, restricted, live; std::atomic epoch{0}; std::mutex waitLock; std::condition_variable wake; bool positioned = false; std::unique_ptr buffered; Input(std::atomic &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((Clock::now() + std::chrono::seconds(15)).time_since_epoch()).count(); deadline = std::min(ioDeadline, operationDeadline.load()); } bool interrupted() const { const auto now = std::chrono::duration_cast(Clock::now().time_since_epoch()).count(); return cancel || now >= operationDeadline.load() || (!buffered && now >= deadline.load()); } static int interrupt(void *opaque) { auto &in = *static_cast(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(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 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(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 bytes; }; struct Mux { AVFormatContext *format = nullptr; AVIOContext *io = nullptr; std::vector bytes; Mux() { check(avformat_alloc_output_context2(&format, nullptr, "mpegts", nullptr)); auto *buffer = static_cast(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(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 continuous; int64_t audioKeepFrom = AV_NOPTS_VALUE, audioKeepUntil = INT64_MAX; OutputTrack(Track &track, Mux &output, bool live) : source(track), mux(&output), continuous(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)); } firstVideo = true; } 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); if (fifo && !continuous && audioKeepFrom != AV_NOPTS_VALUE && (p->pts < audioKeepFrom || p->pts >= audioKeepUntil)) { 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(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 = continuous ? start : aacPreroll(start); const double limit = continuous ? end : double(aacBoundary(end)) / 48000; if (!continuous) { audioKeepFrom = aacBoundary(start); audioKeepUntil = int64_t(std::llround(end * 48000)); } 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(f->extended_data), f->nb_samples); check(count); int skip = continuous && audioPTS != AV_NOPTS_VALUE ? 0 : std::clamp(int(std::llround((begin - time) * 48000)), 0, count); int take = continuous ? count - skip : std::clamp(int(std::llround((limit - time) * 48000)) - skip, 0, 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(data), take) == take); audioFrames(false, end); } } source.nextFrame(); } void finish(double end) { if (!encoder) return; if (continuous) { if (fifo) audioFrames(false, end); } else { if (fifo) audioFrames(true); encoded(nullptr); } } }; struct Pipeline { playback::StreamCacheBudget cacheBudget; std::unique_ptr videoInput, audioInput; Input *audioSource = nullptr; int videoIndex = -1, audioIndex = -1; Clock::time_point deadline; std::unique_ptr video, audio; std::unique_ptr videoOutput, audioOutput; bool live; std::atomic &cancel; double duration = 0; std::vector boundaries; Pipeline(const Source &source, std::atomic &stop, double start) : live(source.live), cancel(stop) { videoInput = std::make_unique(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(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(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(deadline.time_since_epoch()).count(); videoInput->operationDeadline = limit; if (audioInput) audioInput->operationDeadline = limit; } void releaseOperation() { videoInput->operationDeadline = std::numeric_limits::max(); if (audioInput) audioInput->operationDeadline = std::numeric_limits::max(); } void reset(double start, bool aligned = false) { armOperation(); videoOutput.reset(); audioOutput.reset(); video.reset(); audio.reset(); // Fresh codecs retain independent IDRs/AAC priming. One recorded network // demuxer atomically rewinds both packet lanes, seeking upstream only on 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(*videoInput, videoIndex, AVMEDIA_TYPE_VIDEO, start, live, aligned); if (audioIndex >= 0) audio = std::make_unique(*audioSource, audioIndex, AVMEDIA_TYPE_AUDIO, start, live); } Segment segment(int64_t id, double start, double end) { if (live) armOperation(); Mux mux; if (video) { videoOutput.reset(); videoOutput = std::make_unique(*video, mux, false); } if (audio) { if (audioOutput) audioOutput->beginSegment(mux); else audioOutput = std::make_unique(*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); 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(end); if (audioOutput) audioOutput->finish(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); 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 cancelled{false}; std::mutex lock; std::mutex vodLock; std::unique_ptr vodPipeline; std::condition_variable wake; int listener = -1; std::vector clients; std::array servers; std::thread producer; std::string base, token; double duration = 0; std::vector boundaries; std::deque> segments; std::map>> 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 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(&local), &length) == 0; close(route); if (found) { family = address->ai_family; break; } } require(family != 0 && !cancelled); char name[NI_MAXHOST]; require(getnameinfo(reinterpret_cast(&local), length, name, sizeof(name), nullptr, 0, NI_NUMERICHOST) == 0); if (family == AF_INET) reinterpret_cast(&local)->sin_port = 0; else reinterpret_cast(&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(&local), length) != 0 || listen(server, 8) != 0 || getsockname(server, reinterpret_cast(&local), &length) != 0) { close(server); require(false); } unsigned port = ntohs(family == AF_INET ? reinterpret_cast(&local)->sin_port : reinterpret_cast(&local)->sin6_port); std::array 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) { 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(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 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 independently generated segment resets codec/mux state. 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 vodSegment(int64_t id) { auto cached = [&]() -> std::shared_ptr { 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) vodPipeline->reset(start, !boundaries.empty()); else vodPipeline = std::make_unique(source, cancelled, start); auto segment = std::make_shared(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 &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(header.data()), header.size()); if (!head && end > begin) sendAll(socket, bytes.data() + begin, end - begin); } void serve(int socket) { std::string request; std::array 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 data; const char *type = "video/mp2t"; int status = 200; std::shared_ptr> 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>(*found, &(*found)->bytes); } else if (id >= segmentCount()) status = 404; else { auto segment = vodSegment(id); retained = std::shared_ptr>(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(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(bytes); media->subtitles[id] = std::make_shared>(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