1 defmodule Pleroma.Web.Streamer.State do
5 alias Pleroma.Web.Streamer.StreamerSocket
10 GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__)
13 def add_socket(topic, socket) do
14 GenServer.call(__MODULE__, {:add, topic, socket})
17 def remove_socket(topic, socket) do
18 do_remove_socket(@env, topic, socket)
22 %{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state)
30 def handle_call(:get_state, _from, state) do
31 {:reply, state, state}
34 def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do
35 internal_topic = internal_topic(topic, socket)
36 stream_socket = StreamerSocket.from_socket(socket)
40 |> Map.get(internal_topic, [])
41 |> List.insert_at(0, stream_socket)
44 state = put_in(state, [:sockets, internal_topic], sockets_for_topic)
45 Logger.debug("Got new conn for #{topic}")
46 {:reply, state, state}
49 def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do
50 internal_topic = internal_topic(topic, socket)
51 stream_socket = StreamerSocket.from_socket(socket)
55 |> Map.get(internal_topic, [])
56 |> List.delete(stream_socket)
58 state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic)
59 {:reply, state, state}
62 defp do_remove_socket(:test, _, _) do
66 defp do_remove_socket(_env, topic, socket) do
67 GenServer.call(__MODULE__, {:remove, topic, socket})
70 defp internal_topic(topic, socket)
71 when topic in ~w[user user:notification direct] do
72 "#{topic}:#{socket.assigns[:user].id}"
75 defp internal_topic(topic, _) do