defmodule UrbitEx.Server do alias UrbitEx.{API} use GenServer require Logger @moduledoc """ Main GenServer, called by the UrbitEx module. Starts a GenServer which keeps an Urbit login session as its state, keeping track of the Eyre channel, the actions sent and events received. """ def start_link(options, name \\ :urbit) when is_list(options) do GenServer.start_link(__MODULE__, options, name: name) end # callbacks @impl true def init(args) do [{:url, url}, {:code, code}] = args session = API.init(url, code) |> API.login() |> API.open_channel() |> API.start_sse() {:ok, session} end @impl true def handle_info(%HTTPoison.AsyncStatus{}, session) do {:noreply, session} end @impl true # ignore keep-alive messages def handle_info(%{chunk: "\n"}, session) do {:noreply, session} end @impl true def handle_info(%{chunk: data}, session) do {event_id, message} = data |> parse_stream new_session = %{session | last_sse: event_id, events: [message | session.events]} broadcast(session, message) {:noreply, new_session} end @impl true def handle_info(_message, session) do {:noreply, session} end defp parse_stream(event) do # require IEx # IEx.pry() [id_string, data_string] = event |> String.split("\n") |> Enum.filter(fn x -> String.length(x) > 0 end) [id] = Regex.run(~r(\d), id_string) [_, data] = data_string |> String.split("data: ") {String.to_integer(id), Jason.decode!(data)} end @impl true def handle_call(:get, _from, session) do {:reply, session, session} end @impl true def handle_cast({:subscribe, subscriptions}, session) do session = API.subscribe(session, subscriptions) session = %{session | subscriptions: [subscriptions | session.subscriptions]} {:noreply, session} end @impl true def handle_cast({:consume, pid}, session) do {:noreply, %{session | consumers: [pid | session.consumers]}} end defp broadcast(session, message) do session.consumers |> Enum.each(&send(&1, message)) end end