1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Web.Streamer.State do
9 alias Pleroma.Web.Streamer.StreamerSocket
14 GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__)
17 def add_socket(topic, socket) do
18 GenServer.call(__MODULE__, {:add, topic, socket})
21 def remove_socket(topic, socket) do
22 do_remove_socket(@env, topic, socket)
26 %{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state)
34 def handle_call(:get_state, _from, state) do
35 {:reply, state, state}
38 def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do
39 internal_topic = internal_topic(topic, socket)
40 stream_socket = StreamerSocket.from_socket(socket)
44 |> Map.get(internal_topic, [])
45 |> List.insert_at(0, stream_socket)
48 state = put_in(state, [:sockets, internal_topic], sockets_for_topic)
49 Logger.debug("Got new conn for #{topic}")
50 {:reply, state, state}
53 def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do
54 internal_topic = internal_topic(topic, socket)
55 stream_socket = StreamerSocket.from_socket(socket)
59 |> Map.get(internal_topic, [])
60 |> List.delete(stream_socket)
62 state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic)
63 {:reply, state, state}
66 defp do_remove_socket(:test, _, _) do
70 defp do_remove_socket(_env, topic, socket) do
71 GenServer.call(__MODULE__, {:remove, topic, socket})
74 defp internal_topic(topic, socket)
75 when topic in ~w[user user:notification direct] do
76 "#{topic}:#{socket.assigns[:user].id}"
79 defp internal_topic(topic, _) do