diff --git a/config/config.exs b/config/config.exs index a0176a72d..7514fc572 100644 --- a/config/config.exs +++ b/config/config.exs @@ -888,6 +888,8 @@ config :pleroma, :libre_translate, url: "http://127.0.0.1:5000", api_key: nil +config :pleroma, :queue, module: Pleroma.Broadway + # Import environment specific config. This must remain at the bottom # of this file so it overrides the configuration defined above. import_config "#{Mix.env()}.exs" diff --git a/lib/mix/tasks/pleroma/queue.ex b/lib/mix/tasks/pleroma/queue.ex new file mode 100644 index 000000000..bfcabb973 --- /dev/null +++ b/lib/mix/tasks/pleroma/queue.ex @@ -0,0 +1,18 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2021 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Mix.Tasks.Pleroma.Queue do + use Mix.Task + + import Mix.Pleroma + + def run(["queues"]) do + start_pleroma() + + Pleroma.Config.get([Oban, :queues]) + |> Keyword.keys() + |> Enum.join("\n") + |> shell_info() + end +end diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex index 48a2623ce..a73fead00 100644 --- a/lib/pleroma/application.ex +++ b/lib/pleroma/application.ex @@ -68,11 +68,11 @@ defmodule Pleroma.Application do ] ++ cachex_children() ++ http_children() ++ + queue_children() ++ [ Pleroma.Stats, Pleroma.JobQueueMonitor, {Majic.Pool, [name: Pleroma.MajicPool, pool_size: Config.get([:majic_pool, :size], 2)]}, - {Oban, Config.get(Oban)}, Pleroma.Web.Endpoint ] ++ elasticsearch_children() ++ @@ -267,4 +267,16 @@ defmodule Pleroma.Application do [{Finch, config}] end + + defp queue_children do + queue_module = Config.get([:queue, :module]) + + case queue_module do + Oban -> + [{Oban, Config.get(Oban)}] + + Pleroma.Broadway -> + Pleroma.Broadway.children() + end + end end diff --git a/lib/pleroma/broadway.ex b/lib/pleroma/broadway.ex new file mode 100644 index 000000000..c86ae6836 --- /dev/null +++ b/lib/pleroma/broadway.ex @@ -0,0 +1,93 @@ +defmodule Pleroma.Broadway do + use Broadway + alias Broadway.Message + require Logger + + @queue "akkoma" + + def start_link(_args) do + Broadway.start_link(__MODULE__, + name: __MODULE__, + producer: [ + module: + {BroadwayRabbitMQ.Producer, + queue: @queue, + declare: [ + durable: true, + auto_delete: false + ], + on_failure: :reject_and_requeue + } + ], + processors: [ + default: [ + concurrency: 10 + ] + ], + batchers: [ + default: [ + batch_size: 10, + batch_timeout: 100, + concurrency: 10 + ] + ] + ) + end + + @impl true + def handle_message(_, %Message{data: data} = message, _) do + with {:ok, data} <- Jason.decode(data), + {module, data} <- Map.pop(data, "__module__"), + module <- String.to_existing_atom(module), + :ok <- perform_message(module, data) do + Logger.debug("Received message: #{inspect(data)}") + message + else + err -> + IO.inspect(err) + Message.failed(message, err) + end + end + + defp perform_message(module, args) do + IO.inspect(args) + case module.perform(%Oban.Job{args: args}) do + :ok -> + :ok + + {:ok, _} -> + :ok + + err -> + err + end + end + + @impl true + def handle_batch(_, batch, _, _) do + batch + end + + @impl true + def handle_failed(messages, _) do + Logger.error("Failed messages: #{inspect(messages)}") + messages + end + + def topics do + Pleroma.Config.get([Oban, :queues]) + |> Keyword.keys() + end + + def children do + [Pleroma.Broadway] + end + + def produce(topic, args) do + IO.puts("Producing message on #{topic}: #{inspect(args)}") + {:ok, connection} = AMQP.Connection.open() + {:ok, channel} = AMQP.Channel.open(connection) + AMQP.Basic.publish(channel, "", @queue, args) + AMQP.Connection.close(connection) + end +end diff --git a/lib/pleroma/workers/publisher_worker.ex b/lib/pleroma/workers/publisher_worker.ex index 545887071..84ba2d179 100644 --- a/lib/pleroma/workers/publisher_worker.ex +++ b/lib/pleroma/workers/publisher_worker.ex @@ -15,6 +15,7 @@ defmodule Pleroma.Workers.PublisherWorker do @impl Oban.Worker def perform(%Job{args: %{"op" => "publish", "activity_id" => activity_id, "object_data" => nil}}) do activity = Activity.get_by_id(activity_id) + IO.inspect(activity) Federator.perform(:publish, activity) end diff --git a/lib/pleroma/workers/worker_helper.ex b/lib/pleroma/workers/worker_helper.ex index 4c0a55774..a55dfc14a 100644 --- a/lib/pleroma/workers/worker_helper.ex +++ b/lib/pleroma/workers/worker_helper.ex @@ -25,30 +25,55 @@ defmodule Pleroma.Workers.WorkerHelper do defmacro __using__(opts) do caller_module = __CALLER__.module queue = Keyword.fetch!(opts, :queue) + queue_system = Config.get([:queue, :module]) - quote do - # Note: `max_attempts` is intended to be overridden in `new/2` call - use Oban.Worker, - queue: unquote(queue), - max_attempts: 1 + case queue_system do + Oban -> + quote do + # Note: `max_attempts` is intended to be overridden in `new/2` call + use Oban.Worker, + queue: unquote(queue), + max_attempts: 1 - alias Oban.Job + alias Oban.Job - def enqueue(op, params, worker_args \\ []) do - params = Map.merge(%{"op" => op}, params) - queue_atom = String.to_atom(unquote(queue)) - worker_args = worker_args ++ WorkerHelper.worker_args(queue_atom) + def enqueue(op, params, worker_args \\ []) do + params = Map.merge(%{"op" => op}, params) + queue_atom = String.to_atom(unquote(queue)) + worker_args = worker_args ++ WorkerHelper.worker_args(queue_atom) - unquote(caller_module) - |> apply(:new, [params, worker_args]) - |> Oban.insert() - end + unquote(caller_module) + |> apply(:new, [params, worker_args]) + |> Oban.insert() + end - @impl Oban.Worker - def timeout(_job) do - queue_atom = String.to_atom(unquote(queue)) - Config.get([:workers, :timeout, queue_atom], :timer.minutes(1)) - end + @impl Oban.Worker + def timeout(_job) do + queue_atom = String.to_atom(unquote(queue)) + Config.get([:workers, :timeout, queue_atom], :timer.minutes(1)) + end + end + + Pleroma.Broadway -> + quote do + @topic unquote(queue) + use Oban.Worker, + queue: unquote(queue), + max_attempts: 1 + + alias Oban.Job + + def enqueue(op, params, worker_args \\ []) do + worker = to_string(__MODULE__) + + params = + params + |> Map.put("__module__", worker) + |> Map.put("op", op) + + Pleroma.Broadway.produce(unquote(queue), Jason.encode!(params)) + end + end end end end diff --git a/mix.exs b/mix.exs index 00a250002..2a85a379a 100644 --- a/mix.exs +++ b/mix.exs @@ -190,6 +190,8 @@ defmodule Pleroma.Mixfile do git: "https://akkoma.dev/AkkomaGang/mfm-parser.git", ref: "912fba81152d4d572e457fd5427f9875b2bc3dbe"}, {:poison, ">= 0.0.0"}, + {:broadway, "~> 1.0"}, + {:broadway_rabbitmq, "~> 0.7"}, ## dev & test {:ex_doc, "~> 0.22", only: :dev, runtime: false},