X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Ffederator%2Ffederator.ex;fp=lib%2Fpleroma%2Fweb%2Ffederator%2Ffederator.ex;h=a1f6373a45b0b2d9a3a58249ef12976516cc5d3c;hb=627e5a0a4992cc19fc65a7e93a09c470c8e2bf33;hp=0644f8d0a4536c45b7c74bb9ed58a0c1a34c8074;hpb=d6ab701a14f7c9fb4d59953648c425e04725fc62;p=akkoma diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex index 0644f8d0a..a1f6373a4 100644 --- a/lib/pleroma/web/federator/federator.ex +++ b/lib/pleroma/web/federator/federator.ex @@ -1,55 +1,85 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2019 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + defmodule Pleroma.Web.Federator do - use GenServer - alias Pleroma.User alias Pleroma.Activity - alias Pleroma.Object.Containment - alias Pleroma.Web.{WebFinger, Websub} - alias Pleroma.Web.Federator.RetryQueue + alias Pleroma.User alias Pleroma.Web.ActivityPub.ActivityPub alias Pleroma.Web.ActivityPub.Relay alias Pleroma.Web.ActivityPub.Transmogrifier alias Pleroma.Web.ActivityPub.Utils + alias Pleroma.Web.ActivityPub.Visibility + alias Pleroma.Web.Federator.RetryQueue alias Pleroma.Web.OStatus + alias Pleroma.Object.Containment + alias Pleroma.Web.Salmon + alias Pleroma.Web.WebFinger + alias Pleroma.Web.Websub + require Logger @websub Application.get_env(:pleroma, :websub) @ostatus Application.get_env(:pleroma, :ostatus) - @httpoison Application.get_env(:pleroma, :httpoison) - @max_jobs 20 - def init(args) do - {:ok, args} + def init do + # 1 minute + Process.sleep(1000 * 60) + refresh_subscriptions() end - def start_link do - spawn(fn -> - # 1 minute - Process.sleep(1000 * 60 * 1) - enqueue(:refresh_subscriptions, nil) - end) + # Client API + + def incoming_doc(doc) do + PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc]) + end + + def incoming_ap_doc(params) do + PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params]) + end + + def publish(activity, priority \\ 1) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish, activity], priority) + end + + def publish_single_ap(params) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_ap, params]) + end + + def publish_single_websub(websub) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_websub, websub]) + end + + def verify_websub(websub) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:verify_websub, websub]) + end + + def request_subscription(sub) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:request_subscription, sub]) + end + + def refresh_subscriptions do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:refresh_subscriptions]) + end - GenServer.start_link( - __MODULE__, - %{ - in: {:sets.new(), []}, - out: {:sets.new(), []} - }, - name: __MODULE__ - ) + def publish_single_salmon(params) do + PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_salmon, params]) end - def handle(:refresh_subscriptions, _) do + # Job Worker Callbacks + + def perform(:refresh_subscriptions) do Logger.debug("Federator running refresh subscriptions") Websub.refresh_subscriptions() spawn(fn -> # 6 hours Process.sleep(1000 * 60 * 60 * 6) - enqueue(:refresh_subscriptions, nil) + refresh_subscriptions() end) end - def handle(:request_subscription, websub) do + def perform(:request_subscription, websub) do Logger.debug("Refreshing #{websub.topic}") with {:ok, websub} <- Websub.request_subscription(websub) do @@ -59,13 +89,13 @@ defmodule Pleroma.Web.Federator do end end - def handle(:publish, activity) do + def perform(:publish, activity) do Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end) with actor when not is_nil(actor) <- User.get_cached_by_ap_id(activity.data["actor"]) do {:ok, actor} = WebFinger.ensure_keys_present(actor) - if ActivityPub.is_public?(activity) do + if Visibility.is_public?(activity) do if OStatus.is_representable?(activity) do Logger.info(fn -> "Sending #{activity.data["id"]} out via WebSub" end) Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity) @@ -85,7 +115,7 @@ defmodule Pleroma.Web.Federator do end end - def handle(:verify_websub, websub) do + def perform(:verify_websub, websub) do Logger.debug(fn -> "Running WebSub verification for #{websub.id} (#{websub.topic}, #{websub.callback})" end) @@ -93,12 +123,12 @@ defmodule Pleroma.Web.Federator do @websub.verify(websub) end - def handle(:incoming_doc, doc) do + def perform(:incoming_doc, doc) do Logger.info("Got document, trying to parse") @ostatus.handle_incoming(doc) end - def handle(:incoming_ap_doc, params) do + def perform(:incoming_ap_doc, params) do Logger.info("Handling incoming AP activity") params = Utils.normalize_params(params) @@ -123,7 +153,11 @@ defmodule Pleroma.Web.Federator do end end - def handle(:publish_single_ap, params) do + def perform(:publish_single_salmon, params) do + Salmon.send_to_user(params) + end + + def perform(:publish_single_ap, params) do case ActivityPub.publish_one(params) do {:ok, _} -> :ok @@ -133,9 +167,9 @@ defmodule Pleroma.Web.Federator do end end - def handle( + def perform( :publish_single_websub, - %{xml: xml, topic: topic, callback: callback, secret: secret} = params + %{xml: _xml, topic: _topic, callback: _callback, secret: _secret} = params ) do case Websub.publish_one(params) do {:ok, _} -> @@ -146,75 +180,11 @@ defmodule Pleroma.Web.Federator do end end - def handle(type, _) do + def perform(type, _) do Logger.debug(fn -> "Unknown task: #{type}" end) {:error, "Don't know what to do with this"} end - if Mix.env() == :test do - def enqueue(type, payload, priority \\ 1) do - if Pleroma.Config.get([:instance, :federating]) do - handle(type, payload) - end - end - else - def enqueue(type, payload, priority \\ 1) do - if Pleroma.Config.get([:instance, :federating]) do - GenServer.cast(__MODULE__, {:enqueue, type, payload, priority}) - end - end - end - - def maybe_start_job(running_jobs, queue) do - if :sets.size(running_jobs) < @max_jobs && queue != [] do - {{type, payload}, queue} = queue_pop(queue) - {:ok, pid} = Task.start(fn -> handle(type, payload) end) - mref = Process.monitor(pid) - {:sets.add_element(mref, running_jobs), queue} - else - {running_jobs, queue} - end - end - - def handle_cast({:enqueue, type, payload, _priority}, state) - when type in [:incoming_doc, :incoming_ap_doc] do - %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state - i_queue = enqueue_sorted(i_queue, {type, payload}, 1) - {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue) - {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}} - end - - def handle_cast({:enqueue, type, payload, _priority}, state) do - %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state - o_queue = enqueue_sorted(o_queue, {type, payload}, 1) - {o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue) - {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}} - end - - def handle_cast(m, state) do - IO.inspect("Unknown: #{inspect(m)}, #{inspect(state)}") - {:noreply, state} - end - - def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do - %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state - i_running_jobs = :sets.del_element(ref, i_running_jobs) - o_running_jobs = :sets.del_element(ref, o_running_jobs) - {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue) - {o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue) - - {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}} - end - - def enqueue_sorted(queue, element, priority) do - [%{item: element, priority: priority} | queue] - |> Enum.sort_by(fn %{priority: priority} -> priority end) - end - - def queue_pop([%{item: element} | queue]) do - {element, queue} - end - def ap_enabled_actor(id) do user = User.get_by_ap_id(id)