Emiliano Frascaro
a87cda156b
Metti in pausa ferma RTMP lato app mantenendo la camera aperta, MediaMTX manda offline.mp4 agli spettatori e Chiudi termina definitivamente. Co-authored-by: Cursor <cursoragent@cursor.com>
98 lines
2.6 KiB
Ruby
98 lines
2.6 KiB
Ruby
module Webhooks
|
|
class MediamtxHandler
|
|
def initialize(event:, session_id:, metadata: {})
|
|
@event = event
|
|
@session_id = session_id
|
|
@metadata = metadata
|
|
end
|
|
|
|
def call
|
|
session = StreamSession.find_by(id: @session_id)
|
|
return unless session
|
|
|
|
case @event
|
|
when "connect"
|
|
handle_connect(session)
|
|
when "disconnect"
|
|
handle_disconnect(session)
|
|
when "ready"
|
|
handle_ready(session)
|
|
end
|
|
end
|
|
|
|
private
|
|
|
|
def handle_connect(session)
|
|
return if session.paused?
|
|
|
|
return if session.live? && session.stream_events.where(event_type: "connected").exists?
|
|
|
|
if session.reconnecting?
|
|
session.reconnect! if session.may_reconnect?
|
|
cancel_timeout(session)
|
|
log(session, "reconnected")
|
|
broadcast(session, "reconnected")
|
|
elsif session.connecting? || session.idle?
|
|
session.go_live! if session.may_go_live?
|
|
log(session, "connected")
|
|
broadcast(session, "connected")
|
|
end
|
|
end
|
|
|
|
def handle_disconnect(session)
|
|
return if session.paused?
|
|
return unless session.live? || session.connecting?
|
|
|
|
if session.connecting?
|
|
session.update!(status: "reconnecting")
|
|
elsif session.may_lose_connection?
|
|
session.lose_connection!
|
|
end
|
|
log(session, "disconnected", @metadata)
|
|
broadcast(session, "disconnected")
|
|
schedule_timeout(session)
|
|
end
|
|
|
|
def handle_ready(session)
|
|
log(session, "connected", { ready: true })
|
|
end
|
|
|
|
def schedule_timeout(session)
|
|
job = DisconnectionTimeoutJob.perform_in(
|
|
MatchLiveTv.reconnect_timeout_seconds.seconds,
|
|
session.id
|
|
)
|
|
session.update!(timeout_job_id: job)
|
|
end
|
|
|
|
def cancel_timeout(session)
|
|
return if session.timeout_job_id.blank?
|
|
|
|
Sidekiq::ScheduledSet.new.find_job(session.timeout_job_id)&.delete
|
|
session.update!(timeout_job_id: nil)
|
|
end
|
|
|
|
def log(session, type, meta = {})
|
|
return if duplicate_event?(session, type, meta)
|
|
|
|
session.stream_events.create!(
|
|
event_type: type,
|
|
metadata: meta.merge(source: "mediamtx"),
|
|
occurred_at: Time.current
|
|
)
|
|
end
|
|
|
|
def duplicate_event?(session, type, meta)
|
|
session.stream_events
|
|
.where(event_type: type)
|
|
.where("occurred_at > ?", 2.seconds.ago)
|
|
.where(metadata: meta.merge(source: "mediamtx"))
|
|
.exists?
|
|
end
|
|
|
|
def broadcast(session, event)
|
|
SessionChannel.broadcast_message(session, { type: "stream_event", event: event })
|
|
end
|
|
end
|
|
end
|