git.squeep.com Git - akkoma/blob - lib/pleroma/web/streamer/state.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.Streamer.State do
   6   use GenServer
   7   require Logger
   8
   9   alias Pleroma.Web.Streamer.StreamerSocket
  10
  11   @env Mix.env()
  12
  13   def start_link(_) do
  14     GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__)
  15   end
  16
  17   def add_socket(topic, socket) do
  18     GenServer.call(__MODULE__, {:add, topic, socket})
  19   end
  20
  21   def remove_socket(topic, socket) do
  22     do_remove_socket(@env, topic, socket)
  23   end
  24
  25   def get_sockets do
  26     %{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state)
  27     stream_sockets
  28   end
  29
  30   def init(init_arg) do
  31     {:ok, init_arg}
  32   end
  33
  34   def handle_call(:get_state, _from, state) do
  35     {:reply, state, state}
  36   end
  37
  38   def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do
  39     stream_socket = StreamerSocket.from_socket(socket)
  40
  41     sockets_for_topic =
  42       sockets
  43       |> Map.get(topic, [])
  44       |> List.insert_at(0, stream_socket)
  45       |> Enum.uniq()
  46
  47     state = put_in(state, [:sockets, topic], sockets_for_topic)
  48     Logger.debug("Got new conn for #{topic}")
  49     {:reply, state, state}
  50   end
  51
  52   def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do
  53     stream_socket = StreamerSocket.from_socket(socket)
  54
  55     sockets_for_topic =
  56       sockets
  57       |> Map.get(topic, [])
  58       |> List.delete(stream_socket)
  59
  60     state = Kernel.put_in(state, [:sockets, topic], sockets_for_topic)
  61     {:reply, state, state}
  62   end
  63
  64   defp do_remove_socket(:test, _, _) do
  65     :ok
  66   end
  67
  68   defp do_remove_socket(_env, topic, socket) do
  69     GenServer.call(__MODULE__, {:remove, topic, socket})
  70   end
  71 end