X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Fstreamer.ex;h=3136b1b9d352b7cbe5564c0b6989224089a52a2a;hb=1b1e92866742e75de84201b079ffee48c769886e;hp=3a7b9174322a72c8b68d720dd69830ac852e69ab;hpb=08bc31674218cd7ce634b008b7766b48e49c52e3;p=akkoma diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex index 3a7b91743..3136b1b9d 100644 --- a/lib/pleroma/web/streamer.ex +++ b/lib/pleroma/web/streamer.ex @@ -1,13 +1,16 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + defmodule Pleroma.Web.Streamer do use GenServer require Logger - import Plug.Conn + alias Pleroma.{User, Notification, Activity, Object, Repo} + alias Pleroma.Web.ActivityPub.ActivityPub + + @keepalive_interval :timer.seconds(30) def start_link do - spawn(fn -> - Process.sleep(1000 * 30) # 30 seconds - GenServer.cast(__MODULE__, %{action: :ping}) - end) GenServer.start_link(__MODULE__, %{}, name: __MODULE__) end @@ -23,54 +26,204 @@ defmodule Pleroma.Web.Streamer do GenServer.cast(__MODULE__, %{action: :stream, topic: topic, item: item}) end + def init(args) do + spawn(fn -> + # 30 seconds + Process.sleep(@keepalive_interval) + GenServer.cast(__MODULE__, %{action: :ping}) + end) + + {:ok, args} + end + def handle_cast(%{action: :ping}, topics) do Map.values(topics) - |> List.flatten - |> Enum.each(fn (socket) -> + |> List.flatten() + |> Enum.each(fn socket -> Logger.debug("Sending keepalive ping") - send socket.transport_pid, {:text, ""} + send(socket.transport_pid, {:text, ""}) end) + spawn(fn -> - Process.sleep(1000 * 30) # 30 seconds + # 30 seconds + Process.sleep(@keepalive_interval) GenServer.cast(__MODULE__, %{action: :ping}) end) + + {: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 + # filter the recipient list if the activity is not public, see #270. + recipient_lists = + case ActivityPub.is_public?(item) do + true -> + Pleroma.List.get_lists_from_activity(item) + + _ -> + Pleroma.List.get_lists_from_activity(item) + |> Enum.filter(fn list -> + owner = Repo.get(User, list.user_id) + + ActivityPub.visible_for_user?(item, owner) + end) + end + + recipient_topics = + recipient_lists + |> 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}" + + Enum.each(topics[topic] || [], fn socket -> + json = + %{ + event: "notification", + payload: + Pleroma.Web.MastodonAPI.MastodonAPIController.render_notification( + socket.assigns["user"], + item + ) + |> Jason.encode!() + } + |> Jason.encode!() + + send(socket.transport_pid, {:text, json}) + end) + + {:noreply, topics} + end + + def handle_cast(%{action: :stream, topic: "user", item: item}, topics) do + Logger.debug("Trying to push to users") + + recipient_topics = + User.get_recipients_from_activity(item) + |> Enum.map(fn %{id: id} -> "user:#{id}" end) + + Enum.each(recipient_topics, fn topic -> + push_to_socket(topics, topic, item) + end) + {:noreply, topics} end def handle_cast(%{action: :stream, topic: topic, item: item}, topics) do Logger.debug("Trying to push to #{topic}") Logger.debug("Pushing item to #{topic}") - Enum.each(topics[topic] || [], fn (socket) -> - json = %{ - event: "update", - payload: Pleroma.Web.MastodonAPI.StatusView.render("status.json", activity: item) |> Poison.encode! - } |> Poison.encode! - - send socket.transport_pid, {:text, json} - end) + push_to_socket(topics, topic, item) {:noreply, topics} end def handle_cast(%{action: :add, topic: topic, socket: socket}, sockets) do + topic = internal_topic(topic, socket) sockets_for_topic = sockets[topic] || [] sockets_for_topic = Enum.uniq([socket | sockets_for_topic]) sockets = Map.put(sockets, topic, sockets_for_topic) Logger.debug("Got new conn for #{topic}") - IO.inspect(sockets) {:noreply, sockets} end def handle_cast(%{action: :remove, topic: topic, socket: socket}, sockets) do + topic = internal_topic(topic, socket) sockets_for_topic = sockets[topic] || [] sockets_for_topic = List.delete(sockets_for_topic, socket) sockets = Map.put(sockets, topic, sockets_for_topic) Logger.debug("Removed conn for #{topic}") - IO.inspect(sockets) {:noreply, sockets} end def handle_cast(m, state) do - IO.inspect("Unknown: #{inspect(m)}, #{inspect(state)}") + Logger.info("Unknown: #{inspect(m)}, #{inspect(state)}") {:noreply, state} end + + defp represent_update(%Activity{} = activity, %User{} = user) do + %{ + event: "update", + payload: + Pleroma.Web.MastodonAPI.StatusView.render( + "status.json", + activity: activity, + for: user + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + + defp represent_update(%Activity{} = activity) do + %{ + event: "update", + payload: + Pleroma.Web.MastodonAPI.StatusView.render( + "status.json", + activity: activity + ) + |> Jason.encode!() + } + |> Jason.encode!() + end + + def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do + Enum.each(topics[topic] || [], fn socket -> + # Get the current user so we have up-to-date blocks etc. + if socket.assigns[:user] do + user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id) + blocks = user.info.blocks || [] + + parent = Object.normalize(item.data["object"]) + + unless is_nil(parent) or item.actor in blocks or parent.data["actor"] in blocks do + send(socket.transport_pid, {:text, represent_update(item, user)}) + end + else + send(socket.transport_pid, {:text, represent_update(item)}) + end + end) + end + + def push_to_socket(topics, topic, item) do + Enum.each(topics[topic] || [], fn socket -> + # Get the current user so we have up-to-date blocks etc. + if socket.assigns[:user] do + user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id) + blocks = user.info.blocks || [] + + unless item.actor in blocks do + send(socket.transport_pid, {:text, represent_update(item, user)}) + end + else + send(socket.transport_pid, {:text, represent_update(item)}) + end + end) + end + + defp internal_topic(topic, socket) when topic in ~w[user direct] do + "#{topic}:#{socket.assigns[:user].id}" + end + + defp internal_topic(topic, _), do: topic end