Какое поведение OTP следует использовать для "бесконечного" повторения задач?

Я хочу многократно повторять одну и ту же последовательность операций снова и снова рядом с приложением Phoenix (без сбоев в работе всего веб-приложения, если что-то тормозит у работника, конечно) и на самом деле не знаю, я должен использовать GenServer, Elixir Tasks, Агент или что-то совершенно другое, о котором я до сих пор не думал.

Когда я запускаю приложение Phoenix, рабочий тоже должен начинать, что периодически вытягивает некоторые значения последовательного соединения, транслирует их по каналу Phoenix, собирает их до тех пор, пока @save_interval не будет достигнуто, а затем вычислит медианные трансляции, которые медиана через другой канал и записывает его в InfluxDB. Прямо сейчас у меня есть что-то вроде работы:

def do_your_thing(serial_pid) do
  Stream.interval(@interval_live)
    |> get_new_values_from_serial(serial_pid)
    |> broadcast!("live-channel:#{@name}")
    |> Enum.take(div(@interval_save, @interval_live))
    |> calculate_medians()
    |> broadcast!("update-channel:#{@name}")
    |> write_to_database()

  do_your_thing(serial_pid) # repeat
end

Я только начинаю понимать все, что есть OTP, и надеюсь, что кто-то из вас может помочь мне наткнуться в правильном направлении здесь.

Ответы

Ответ 1

Вам следует использовать GenServer, который отправляет сообщения через x секунд (60 секунд в примере ниже):

defmodule MyApp.Worker do
  use GenServer

  def start_link() do
    GenServer.start_link(__MODULE__, [])
  end

  def init([]) do
    schedule_work()
    {:ok, []}
  end

  def handle_info(:work, state) do
    state = do_work(state)
    schedule_work()
    {:noreply, state}
  end

  defp do_work(state) do
    # Do your work here and return state
  end

  defp schedule_work do
    Process.send_after(self(), :work, 60_000)
  end
end