X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fweb%2Ffederator%2Ffederator.ex;h=7df75aca602264b13a49e6d88705b96d8575dc9b;hb=305d2194136d4560e02c110d528164034d3727b2;hp=468959a65f34883b55d07593d149b043374195eb;hpb=06c4935c11fc58f5ede18706bc46dcc502e64052;p=akkoma

diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
index 468959a65..7df75aca6 100644
--- a/lib/pleroma/web/federator/federator.ex
+++ b/lib/pleroma/web/federator/federator.ex
@@ -3,8 +3,6 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Federator do
-  use GenServer
-
   alias Pleroma.Activity
   alias Pleroma.User
   alias Pleroma.Web.WebFinger
@@ -16,45 +14,71 @@ defmodule Pleroma.Web.Federator do
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.Federator.RetryQueue
   alias Pleroma.Web.OStatus
+  alias Pleroma.Jobs
 
   require Logger
 
   @websub Application.get_env(:pleroma, :websub)
   @ostatus Application.get_env(:pleroma, :ostatus)
 
-  def init(args) do
-    {:ok, args}
+  def init() do
+    # 1 minute
+    Process.sleep(1000 * 60 * 1)
+    refresh_subscriptions()
   end
 
-  def start_link do
-    spawn(fn ->
-      # 1 minute
-      Process.sleep(1000 * 60)
-      enqueue(:refresh_subscriptions, nil)
-    end)
+  # Client API
+
+  def incoming_doc(doc) do
+    Jobs.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc])
+  end
+
+  def incoming_ap_doc(params) do
+    Jobs.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params])
+  end
+
+  def publish(activity, priority \\ 1) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:publish, activity], priority)
+  end
+
+  def publish_single_ap(params) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:publish_single_ap, params])
+  end
 
-    GenServer.start_link(
-      __MODULE__,
-      %{
-        in: {:sets.new(), []},
-        out: {:sets.new(), []}
-      },
-      name: __MODULE__
-    )
+  def publish_single_websub(websub) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:publish_single_websub, websub])
   end
 
-  def handle(:refresh_subscriptions, _) do
+  def verify_websub(websub) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:verify_websub, websub])
+  end
+
+  def request_subscription(sub) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:request_subscription, sub])
+  end
+
+  def refresh_subscriptions() do
+    Jobs.enqueue(:federator_out, __MODULE__, [:refresh_subscriptions])
+  end
+
+  def publish_single_salmon(params) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:publish_single_salmon, params])
+  end
+
+  # 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
@@ -64,7 +88,7 @@ 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
@@ -90,7 +114,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)
@@ -98,12 +122,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)
@@ -128,11 +152,11 @@ defmodule Pleroma.Web.Federator do
     end
   end
 
-  def handle(:publish_single_salmon, params) do
+  def perform(:publish_single_salmon, params) do
     Salmon.send_to_user(params)
   end
 
-  def handle(:publish_single_ap, params) do
+  def perform(:publish_single_ap, params) do
     case ActivityPub.publish_one(params) do
       {:ok, _} ->
         :ok
@@ -142,7 +166,7 @@ defmodule Pleroma.Web.Federator do
     end
   end
 
-  def handle(
+  def perform(
         :publish_single_websub,
         %{xml: _xml, topic: _topic, callback: _callback, secret: _secret} = params
       ) do
@@ -155,74 +179,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) < Pleroma.Config.get([__MODULE__, :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(_, state) do
-    {: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)