X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Fstreamer.ex;h=0c9e954854a6ca49f581a69ffa8a59b13a29ac75;hb=4894b88b1b40705e8d5236d4b417ce2791e3b8fd;hp=3375af8414acfc742716d1e34955ead8d2f3102c;hpb=6c2903d9a175cfbf3785d5c1a43e6fcac6b0e9f5;p=akkoma diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex index 3375af841..0c9e95485 100644 --- a/lib/pleroma/web/streamer.ex +++ b/lib/pleroma/web/streamer.ex @@ -3,6 +3,10 @@ defmodule Pleroma.Web.Streamer do require Logger alias Pleroma.{User, Notification} + def init(args) do + {:ok, args} + end + def start_link do spawn(fn -> # 30 seconds @@ -42,6 +46,32 @@ defmodule Pleroma.Web.Streamer do {:noreply, topics} end + def handle_cast(%{action: :stream, topic: "direct", item: item}, topics) do + recipient_topics = + User.get_recipients_from_activity(item) + |> Enum.map(fn %{id: id} -> "direct:#{id}" end) + + Enum.each(recipient_topics || [], fn user_topic -> + Logger.debug("Trying to push direct message to #{user_topic}\n\n") + push_to_socket(topics, user_topic, item) + end) + + {:noreply, topics} + end + + def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do + recipient_topics = + Pleroma.List.get_lists_from_activity(item) + |> Enum.map(fn %{id: id} -> "list:#{id}" end) + + Enum.each(recipient_topics || [], fn list_topic -> + Logger.debug("Trying to push message to #{list_topic}\n\n") + push_to_socket(topics, list_topic, item) + end) + + {:noreply, topics} + end + def handle_cast(%{action: :stream, topic: "user", item: %Notification{} = item}, topics) do topic = "user:#{item.user_id}" @@ -133,8 +163,8 @@ defmodule Pleroma.Web.Streamer do end) end - defp internal_topic("user", socket) do - "user:#{socket.assigns[:user].id}" + defp internal_topic(topic, socket) when topic in ~w[user direct] do + "#{topic}:#{socket.assigns[:user].id}" end defp internal_topic(topic, _), do: topic