Handle incoming items through the queue.

lib/pleroma/web/federator/federator.ex

@@ -5,6 +5,7 @@ defmodule Pleroma.Web.Federator do
   require Logger
 
   @websub Application.get_env(:pleroma, :websub)
+  @ostatus Application.get_env(:pleroma, :ostatus)
   @max_jobs 10
 
   def start_link do
@@ -28,6 +29,11 @@ def handle(:verify_websub, websub) do
     @websub.verify(websub)
   end
 
+  def handle(:incoming_doc, doc) do
+    Logger.debug("Got document, trying to parse")
+    @ostatus.handle_incoming(doc)
+  end
+
   def handle(type, payload) do
     Logger.debug(fn -> "Unknown task: #{type}" end)
     {:error, "Don't know what do do with this"}

lib/pleroma/web/ostatus/ostatus_controller.ex

@@ -4,7 +4,7 @@ defmodule Pleroma.Web.OStatus.OStatusController do
   alias Pleroma.{User, Activity}
   alias Pleroma.Web.OStatus.{FeedRepresenter, ActivityRepresenter}
   alias Pleroma.Repo
-  alias Pleroma.Web.OStatus
+  alias Pleroma.Web.{OStatus, Federator}
   import Ecto.Query
 
   def feed_redirect(conn, %{"nickname" => nickname}) do
@@ -37,7 +37,7 @@ def salmon_incoming(conn, params) do
     {:ok, magic_key} = Pleroma.Web.Salmon.fetch_magic_key(body)
     {:ok, doc} = Pleroma.Web.Salmon.decode_and_validate(magic_key, body)
 
-    Pleroma.Web.OStatus.handle_incoming(doc)
+    Federator.enqueue(:incoming_doc, doc)
 
     conn
     |> send_resp(200, "")

lib/pleroma/web/websub/websub_controller.ex

@@ -1,12 +1,10 @@
 defmodule Pleroma.Web.Websub.WebsubController do
   use Pleroma.Web, :controller
   alias Pleroma.{Repo, User}
-  alias Pleroma.Web.Websub
+  alias Pleroma.Web.{Websub, Federator}
   alias Pleroma.Web.Websub.WebsubClientSubscription
   require Logger
 
-  @ostatus Application.get_env(:pleroma, :ostatus)
-
   def websub_subscription_request(conn, %{"nickname" => nickname} = params) do
     user = User.get_cached_by_nickname(nickname)
 
@@ -38,7 +36,7 @@ def websub_incoming(conn, %{"id" => id}) do
          %WebsubClientSubscription{} = websub <- Repo.get(WebsubClientSubscription, id),
          {:ok, body, _conn} = read_body(conn),
          ^signature <- Websub.sign(websub.secret, body) do
-      @ostatus.handle_incoming(body)
+      Federator.enqueue(:incoming_doc, body)
       conn
       |> send_resp(200, "OK")
     else _e ->