MastoAPI: Implement all streaming functions.

This commit is contained in:
Roger Braun 2017-11-16 16:49:51 +01:00
parent 5719f69ae3
commit a743940463
7 changed files with 73 additions and 6 deletions

View file

@ -78,8 +78,9 @@ defmodule Pleroma.Notification do
# TODO move to sql, too. # TODO move to sql, too.
def create_notification(%Activity{} = activity, %User{} = user) do def create_notification(%Activity{} = activity, %User{} = user) do
unless User.blocks?(user, %{ap_id: activity.data["actor"]}) do unless User.blocks?(user, %{ap_id: activity.data["actor"]}) do
notification = %Notification{user_id: user.id, activity_id: activity.id} notification = %Notification{user_id: user.id, activity: activity}
{:ok, notification} = Repo.insert(notification) {:ok, notification} = Repo.insert(notification)
Pleroma.Web.Streamer.stream("user", notification)
notification notification
end end
end end

View file

@ -284,6 +284,17 @@ defmodule Pleroma.User do
Repo.all(query) Repo.all(query)
end end
def get_recipients_from_activity(%Activity{data: %{"to" => to}} = activity) do
query = from u in User,
where: u.local == true
query = from u in query,
where: u.ap_id in ^to,
or_where: fragment("? \\\?| ?", u.following, ^to)
Repo.all(query)
end
def search(query, resolve) do def search(query, resolve) do
if resolve do if resolve do
User.get_or_fetch_by_nickname(query) User.get_or_fetch_by_nickname(query)

View file

@ -24,6 +24,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
:ok <- maybe_federate(activity) do :ok <- maybe_federate(activity) do
if activity.data["type"] == "Create" and Enum.member?(activity.data["to"], "https://www.w3.org/ns/activitystreams#Public") do if activity.data["type"] == "Create" and Enum.member?(activity.data["to"], "https://www.w3.org/ns/activitystreams#Public") do
Pleroma.Web.Streamer.stream("public", activity) Pleroma.Web.Streamer.stream("public", activity)
Pleroma.Web.Streamer.stream("user", activity)
if local do if local do
Pleroma.Web.Streamer.stream("public:local", activity) Pleroma.Web.Streamer.stream("public:local", activity)
end end

View file

@ -595,7 +595,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
json(conn, []) json(conn, [])
end end
defp render_notification(user, %{id: id, activity: activity, inserted_at: created_at} = _params) do def render_notification(user, %{id: id, activity: activity, inserted_at: created_at} = _params) do
actor = User.get_cached_by_ap_id(activity.data["actor"]) actor = User.get_cached_by_ap_id(activity.data["actor"])
created_at = NaiveDateTime.to_iso8601(created_at) created_at = NaiveDateTime.to_iso8601(created_at)
|> String.replace(~r/(\.\d+)?$/, ".000Z", global: false) |> String.replace(~r/(\.\d+)?$/, ".000Z", global: false)

View file

@ -11,7 +11,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonSocket do
with token when not is_nil(token) <- params["access_token"], with token when not is_nil(token) <- params["access_token"],
%Token{user_id: user_id} <- Repo.get_by(Token, token: token), %Token{user_id: user_id} <- Repo.get_by(Token, token: token),
%User{} = user <- Repo.get(User, user_id), %User{} = user <- Repo.get(User, user_id),
stream when stream in ["public", "public:local"] <- params["stream"] do stream when stream in ["public", "public:local", "user"] <- params["stream"] do
socket = socket socket = socket
|> assign(:topic, params["stream"]) |> assign(:topic, params["stream"])
|> assign(:user, user) |> assign(:user, user)

View file

@ -2,6 +2,7 @@ defmodule Pleroma.Web.Streamer do
use GenServer use GenServer
require Logger require Logger
import Plug.Conn import Plug.Conn
alias Pleroma.{User, Notification}
def start_link do def start_link do
spawn(fn -> spawn(fn ->
@ -37,21 +38,55 @@ defmodule Pleroma.Web.Streamer do
{:noreply, topics} {:noreply, topics}
end end
def handle_cast(%{action: :stream, topic: topic, item: item}, topics) do def push_to_socket(topics, topic, item) do
Logger.debug("Trying to push to #{topic}")
Logger.debug("Pushing item to #{topic}")
Enum.each(topics[topic] || [], fn (socket) -> Enum.each(topics[topic] || [], fn (socket) ->
json = %{ json = %{
event: "update", event: "update",
payload: Pleroma.Web.MastodonAPI.StatusView.render("status.json", activity: item, for: socket.assigns[:user]) |> Poison.encode! payload: Pleroma.Web.MastodonAPI.StatusView.render("status.json", activity: item, for: socket.assigns[:user]) |> Poison.encode!
} |> Poison.encode! } |> Poison.encode!
send socket.transport_pid, {:text, json}
end)
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) |> Poison.encode!
} |> Poison.encode!
send socket.transport_pid, {:text, json} send socket.transport_pid, {:text, json}
end) end)
{:noreply, topics} {:noreply, topics}
end 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}")
push_to_socket(topics, topic, item)
{:noreply, topics}
end
defp internal_topic("user", socket) do
"user:#{socket.assigns[:user].id}"
end
defp internal_topic(topic, socket), do: topic
def handle_cast(%{action: :add, topic: topic, socket: socket}, sockets) do def handle_cast(%{action: :add, topic: topic, socket: socket}, sockets) do
topic = internal_topic(topic, socket)
sockets_for_topic = sockets[topic] || [] sockets_for_topic = sockets[topic] || []
sockets_for_topic = Enum.uniq([socket | sockets_for_topic]) sockets_for_topic = Enum.uniq([socket | sockets_for_topic])
sockets = Map.put(sockets, topic, sockets_for_topic) sockets = Map.put(sockets, topic, sockets_for_topic)
@ -61,6 +96,7 @@ defmodule Pleroma.Web.Streamer do
end end
def handle_cast(%{action: :remove, topic: topic, socket: socket}, sockets) do def handle_cast(%{action: :remove, topic: topic, socket: socket}, sockets) do
topic = internal_topic(topic, socket)
sockets_for_topic = sockets[topic] || [] sockets_for_topic = sockets[topic] || []
sockets_for_topic = List.delete(sockets_for_topic, socket) sockets_for_topic = List.delete(sockets_for_topic, socket)
sockets = Map.put(sockets, topic, sockets_for_topic) sockets = Map.put(sockets, topic, sockets_for_topic)

View file

@ -3,6 +3,7 @@ defmodule Pleroma.UserTest do
alias Pleroma.{User, Repo} alias Pleroma.{User, Repo}
alias Pleroma.Web.OStatus alias Pleroma.Web.OStatus
alias Pleroma.Web.Websub.WebsubClientSubscription alias Pleroma.Web.Websub.WebsubClientSubscription
alias Pleroma.Web.CommonAPI
use Pleroma.DataCase use Pleroma.DataCase
import Pleroma.Factory import Pleroma.Factory
@ -296,5 +297,22 @@ defmodule Pleroma.UserTest do
refute User.blocks?(user, blocked_user) refute User.blocks?(user, blocked_user)
end end
end end
test "get recipients from activity" do
actor = insert(:user)
user = insert(:user, local: true)
user_two = insert(:user, local: false)
addressed = insert(:user, local: true)
addressed_remote = insert(:user, local: false)
{:ok, activity} = CommonAPI.post(actor, %{"status" => "hey @#{addressed.nickname} @#{addressed_remote.nickname}"})
assert [addressed] == User.get_recipients_from_activity(activity)
{:ok, user} = User.follow(user, actor)
recipients = User.get_recipients_from_activity(activity)
assert length(recipients) == 2
assert user in recipients
assert addressed in recipients
end
end end