self-hosted/algora-tv
1
0
Fork 0
mirror of https://github.com/algora-io/tv.git synced 2024-11-26 01:00:20 +02:00
Code Releases Activity

implement retry & restart with exponential backoff to handle websocket disconnections (#64)

Browse Source
This commit is contained in:
Zafer Cesur 2024-08-22 23:45:24 +03:00 committed by GitHub
parent ad78fa402c
commit 33aaadebb0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 46 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
lib/algora/restream/websocket.ex
View File

@ -4,8 +4,17 @@ defmodule Algora.Restream.Websocket do
alias Algora.{Accounts, Chat}
def start_link(%{url: url, video: video}) do
WebSockex.start_link(url, __MODULE__, %{url: url, video: video})
def start_link(%{url: url, video: video} = opts) do
retry = opts[:retry] || 0
restart = opts[:restart] || 0
Logger.info("Starting WebSocket: #{video.id} (restart: #{restart}, retry: #{retry})")
WebSockex.start_link(url, __MODULE__, %{
url: url,
video: video,
restart: restart,
retry: retry
})
end
def handle_frame({:text, msg}, state) do
@ -23,9 +32,41 @@ defmodule Algora.Restream.Websocket do
end
end
def handle_disconnect(_reason, state) do
Logger.error("WebSocket disconnected: #{state.video.id}")
{:reconnect, state}
def handle_connect(_conn, state) do
if state.restart == 0 and state.retry == 0 do
Logger.info("WebSocket connected: #{state.video.id}")
else
Logger.info(
"WebSocket reconnected: #{state.video.id} (restart: #{state.restart}, retry: #{state.retry})"
)
end
{:ok, state}
end
def handle_disconnect(reason, state) do
Logger.error("WebSocket disconnected: #{state.video.id} (reason: #{inspect(reason)})")
# latest_video = Library.get_latest_video(Accounts.get_user!(state.video.user_id))
# socket_video = Library.get_video!(state.video.id)
# if socket_video.id == latest_video.id and socket_video.is_live and state.retry < 10 do
if state.retry < 3 do
:timer.sleep(:timer.seconds(min(2 ** state.retry, 60)))
state = %{state | retry: state.retry + 1}
{:reconnect, state}
else
:timer.sleep(:timer.seconds(min(2 ** state.restart, 60)))
state = %{state | restart: state.restart + 1, retry: 0}
Task.Supervisor.start_child(
Algora.TaskSupervisor,
fn -> Algora.Restream.Websocket.start_link(state) end,
restart: :transient
)
{:ok, state}
end
end
defp handle_payload(%{"eventPayload" => %{"contentModifiers" => %{"whisper" => true}}}, state) do