module Streams # Transcodifica path grezzo → path _air con overlay PNG (tabellone + badge) bruciato nel video. class OverlayRelay class Error < StandardError; end REDIS_KEY = "overlay_relay:pid:%s" class << self def start(session) return if session.terminal? stop(session) if pid_for(session.id).present? Streams::Overlay::Refresh.call(session) dir = Streams::Overlay.overlay_dir(session.id) FileUtils.mkdir_p(dir) png = Streams::Overlay.overlay_png_path(session) pipe = Streams::Overlay.overlay_pipe_path(session) ensure_pipe!(pipe) intake = "#{mediamtx_rtmp_url}/#{session.mediamtx_path_name}" output = "#{mediamtx_rtmp_url}/#{session.mediamtx_overlay_path_name}" log_path = log_file(session) FileUtils.mkdir_p(File.dirname(log_path)) fps = output_fps overlay_fps = overlay_input_fps filter = "[1:v]format=rgba[ol];[0:v]fps=#{fps},format=yuv420p[vid];[vid][ol]overlay=0:0:format=auto[vout]" ffmpeg_pid = Process.spawn( "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning", "-fflags", "+genpts+discardcorrupt", "-rw_timeout", "15000000", "-i", intake, "-f", "image2pipe", "-framerate", overlay_fps.to_s, "-i", pipe.to_s, "-filter_complex", filter, "-map", "[vout]", "-map", "0:a?", "-fps_mode", "cfr", "-r", fps.to_s, "-c:v", "libx264", "-preset", x264_preset, "-bf", "0", "-profile:v", "high", "-pix_fmt", "yuv420p", "-b:v", bitrate_for(session), "-maxrate", maxrate_for(session), "-bufsize", bufsize_for(session), "-g", fps.to_s, "-keyint_min", fps.to_s, "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "1", "-f", "flv", output, %i[out err] => log_path, pgroup: true ) spawn_png_feeder!(ffmpeg_pid, pipe, png, log_path) Process.detach(ffmpeg_pid) store_pid(session.id, ffmpeg_pid) OverlayRefreshJob.schedule_chain(session.id) Rails.logger.info("[OverlayRelay] started pid=#{ffmpeg_pid} session=#{session.id}") ffmpeg_pid rescue Errno::ENOENT => e raise Error, "ffmpeg non disponibile: #{e.message}" end def stop(session) pid = pid_for(session.id) return false if pid.blank? terminate_pid(pid) clear_pid(session.id) pipe = Streams::Overlay.overlay_pipe_path(session) FileUtils.rm_f(pipe) if pipe.exist? OverlayRefreshJob.cancel_chain(session.id) Rails.logger.info("[OverlayRelay] stopped pid=#{pid} session=#{session.id}") true end def running?(session_id) pid = pid_for(session_id) return false if pid.blank? return true unless overlay_relay_process_check_local? process_alive?(pid) end # Riavvia il relay se il processo è morto o _air non pubblica (es. feeder bloccato). def ensure_publishing!(session) return if session.terminal? if running?(session.id) && overlay_path_publishing?(session) return end if running?(session.id) misses = redis.incr(overlay_air_miss_key(session.id)).to_i redis.expire(overlay_air_miss_key(session.id), 120) return if misses < 3 redis.del(overlay_air_miss_key(session.id)) Rails.logger.warn("[OverlayRelay] _air non attivo dopo #{misses} tentativi, restart session=#{session.id}") stop(session) else redis.del(overlay_air_miss_key(session.id)) end start(session) rescue Error => e Rails.logger.warn("[OverlayRelay] ensure_publishing session=#{session.id}: #{e.message}") end private def mediamtx_rtmp_url ENV.fetch("MEDIAMTX_INTERNAL_RTMP_URL", "rtmp://mediamtx:1935").chomp("/") end # Ricodifica con overlay: serve più budget della sorgente (generational loss). def output_video_kbps(session) override = ENV["OVERLAY_RELAY_VIDEO_KBPS"].presence&.to_i return override if override&.positive? source_kbps = [(session.target_bitrate.to_i / 1000), 2500].max [(source_kbps * 1.8).to_i, 4500].max end def bitrate_for(session) "#{output_video_kbps(session)}k" end def maxrate_for(session) kbps = (output_video_kbps(session) * 1.15).to_i "#{kbps}k" end def bufsize_for(session) kbps = output_video_kbps(session) * 2 "#{kbps}k" end def output_fps ENV.fetch("OVERLAY_RELAY_FPS", "30").to_i end def x264_preset ENV.fetch("OVERLAY_RELAY_X264_PRESET", "fast") end def overlay_input_fps ENV.fetch("OVERLAY_INPUT_FPS", "2").to_i end def ensure_pipe!(path) FileUtils.rm_f(path) if path.exist? && !path.pipe? system("mkfifo", path.to_s) unless path.exist? raise Error, "impossibile creare fifo overlay #{path}" unless path.pipe? end def spawn_png_feeder!(ffmpeg_pid, pipe, png, log_path) feeder = Rails.root.join("bin/overlay_png_feeder.rb") poll = (1.0 / overlay_input_fps).round(2) Process.spawn( RbConfig.ruby, feeder.to_s, pipe.to_s, png.to_s, poll.to_s, %i[out err] => log_path, pgroup: ffmpeg_pid ) end def overlay_relay_process_check_local? ENV["OVERLAY_RELAY_PROCESS_CHECK"] != "false" end def overlay_path_publishing?(session) info = Mediamtx::Client.new.list_paths.find { |i| i["name"] == session.mediamtx_overlay_path_name } info && (info["online"] || info["ready"] || info["available"]) end def log_file(session) Rails.root.join("log", "overlay_relay_#{session.id}.log").to_s end def redis @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) end def store_pid(session_id, pid) redis.set(format(REDIS_KEY, session_id), pid, ex: 48.hours.to_i) end def clear_pid(session_id) redis.del(format(REDIS_KEY, session_id)) end def pid_for(session_id) redis.get(format(REDIS_KEY, session_id)) end def overlay_air_miss_key(session_id) format("overlay_relay:air_miss:%s", session_id) end def process_alive?(pid) Process.kill(0, pid.to_i) true rescue Errno::ESRCH, Errno::EPERM false end def terminate_pid(pid) Process.kill("TERM", -pid.to_i) sleep 0.5 rescue Errno::ESRCH nil else begin Process.kill("KILL", -pid.to_i) rescue Errno::ESRCH nil end end end end end