147 lines
4.7 KiB
Ruby
147 lines
4.7 KiB
Ruby
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),
|
|
%i[out err] => log_path,
|
|
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
|
|
|
|
# Non usare ffprobe su RTMP live: si blocca e impedisce l'avvio del relay (Sidekiq in hang).
|
|
# Audio: traccia AAC da anullsrc (YouTube la richiede); il mic del telefono resta su _air/HLS.
|
|
def youtube_ffmpeg_args(intake, output)
|
|
[
|
|
"ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning",
|
|
"-fflags", "+genpts+discardcorrupt", "-use_wallclock_as_timestamps", "1",
|
|
"-rw_timeout", "15000000",
|
|
"-i", intake,
|
|
"-f", "lavfi", "-i", "anullsrc=channel_layout=mono:sample_rate=48000",
|
|
"-map", "0:v:0", "-map", "1:a:0",
|
|
"-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
|
|
]
|
|
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
|