module Streams # Relay continuo verso YouTube: legge l'uscita MediaMTX del path (camera o slate alwaysAvailable) # senza interrompere FLV quando il telefono si mette in pausa o perde rete. class YoutubeRelay class Error < StandardError; end REDIS_KEY = "youtube_relay:pid:%s" class << self def start(session) return unless session.platform == "youtube" return if session.stream_key.blank? return if session.terminal? stop(session) if pid_for(session.id).present? log_path = log_file(session) FileUtils.mkdir_p(File.dirname(log_path)) intake = mediamtx_intake_url(session) output = "rtmp://a.rtmp.youtube.com/live2/#{session.stream_key}" # YouTube richiede FLV H.264/AAC stabili; -c copy lascia spesso streamStatus=inactive. pid = Process.spawn( *youtube_ffmpeg_args(intake, output, log_path, with_audio: intake_has_audio?(intake)), pgroup: true ) Process.detach(pid) store_pid(session.id, pid) Rails.logger.info("[YoutubeRelay] started pid=#{pid} session=#{session.id}") schedule_youtube_activate(session) 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) Rails.logger.info("[YoutubeRelay] stopped pid=#{pid} session=#{session.id}") true end def running?(session_id) pid = pid_for(session_id) pid.present? && process_alive?(pid) end # Riavvia il relay se รจ morto (es. avviato prima del segnale su _air). def ensure_publishing!(session) return unless session.platform == "youtube" return if session.terminal? return if session.stream_key.blank? return unless Streams::OverlayRelay.running?(session.id) return if running?(session.id) last_restart = redis.get(restart_debounce_key(session.id)).to_i return if last_restart.positive? && (Time.now.to_i - last_restart) < 20 start(session) redis.set(restart_debounce_key(session.id), Time.now.to_i, ex: 300) rescue Error => e Rails.logger.warn("[YoutubeRelay] ensure_publishing session=#{session.id}: #{e.message}") end private def youtube_ffmpeg_args(intake, output, log_path, with_audio:) base = [ "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning", "-fflags", "+genpts+discardcorrupt", "-use_wallclock_as_timestamps", "1", "-rw_timeout", "15000000", "-i", intake ] encode = youtube_encode_flags(output, log_path) if with_audio base + ["-map", "0:v:0", "-map", "0:a:0"] + encode else # _air senza audio (overlay 0:a?): YouTube richiede comunque AAC. base + [ "-f", "lavfi", "-i", "anullsrc=channel_layout=mono:sample_rate=48000", "-map", "0:v:0", "-map", "1:a:0" ] + encode end end def youtube_encode_flags(output, log_path) [ "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", "-pix_fmt", "yuv420p", "-r", "30", "-g", "30", "-keyint_min", "30", "-sc_threshold", "0", "-b:v", "4000k", "-maxrate", "4000k", "-bufsize", "8000k", "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "1", "-f", "flv", output, %i[out err] => log_path ] end def intake_has_audio?(url) require "open3" stdout, _stderr, status = Open3.capture3( "ffprobe", "-v", "error", "-rw_timeout", "5000000", "-select_streams", "a", "-show_entries", "stream=index", "-of", "csv=p=0", url ) status.success? && stdout.strip.present? rescue StandardError false end def schedule_youtube_activate(session) return if session.youtube_broadcast_id.blank? YoutubeBroadcastActivateJob.schedule_with_retries(session.id) end def mediamtx_intake_url(session) base = ENV.fetch("MEDIAMTX_INTERNAL_RTMP_URL", "rtmp://mediamtx:1935") "#{base.chomp('/')}/#{session.mediamtx_overlay_path_name}" end def log_file(session) Rails.root.join("log", "youtube_relay_#{session.id}.log").to_s end def redis @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) end def restart_debounce_key(session_id) format("youtube_relay:debounce:%s", session_id) 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 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