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) when is_list(options) do GenServer.start_link(__MODULE__, options, name: name) end def start(options, name) when is_list(options) do GenServer.start(__MODULE__, options, name: name) end def die(name \\ :urbit_server) when is_atom(name) do pid = Process.whereis(name) Process.exit(pid, :kill) end def die(pid), do: Process.exit(pid, :kill) # 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 def handle_info(%{chunk: ""}, session) do {:noreply, session} end @impl true def handle_info(%{chunk: data}, session) do parsed_events = data |> parse_stream parsed_events |> Enum.each(fn {_event_id, message} -> broadcast(session, message) end) {last_event_id, last_message} = parsed_events |> Enum.at(-1) new_session = %{session | last_sse: last_event_id, events: [last_message | session.events]} {:noreply, new_session} end @impl true def handle_info(_message, session) do {:noreply, session} end defp parse_stream(event) do event |> String.split("\n") |> Enum.filter(fn x -> String.length(x) > 0 end) # received events can be either single or multiple, which is rather annoying. received_event = event |> String.split("\n") |> Enum.filter(fn x -> String.length(x) > 0 end) received_event |> Enum.chunk_every(2) |> Enum.map(fn [id_string, data_string] -> [id] = Regex.run(~r(\d), id_string) [_, data] = data_string |> String.split("data: ") {String.to_integer(id), Jason.decode!(data)} end) 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 IO.inspect(pid, label: :consooming) {:noreply, %{session | consumers: [pid | session.consumers]}} end defp broadcast(session, message) do session.consumers |> Enum.each(&send(&1, message)) end end