Periodic tasks in a single GenServer -- not everything yet makes sense

I may have misunderstood what you’re trying to do, but this might help.

Assuming that you only wanted to store the user ID for each ask, I used a MapSet. If you want to store additional data with the task (such as a specific URL to poll), you should use a Map (with the key being the user ID, as map keys are unique).

In process_tasks/1 we use Enum.filter/2 to both call a function for each task, and to remove tasks that the external API has indicated should be terminated. The case ExternalAPI.poll(id) do is of course, a contrived example. What’s important is that the anonymous function given to Enum.filter/2 returns false if the task should be removed. (You can give Enum.filter/2 a map too, if you go that route. In that event the anonymous function should take a tuple {user_id, data} representing each key-value pair in the map.)

There’s no need to stop the GenServer when there are no tasks. Erlang processes are very lightweight.

defmodule MyWorker do
  use GenServer

  @tick_interval 30_000

  def start_link, do: GenServer.start_link(__MODULE__, nil, name: __MODULE__)

  def add(id), do: GenServer.cast(__MODULE__, {:add, id})

  def remove(id), do: GenServer.cast(__MODULE__, {:remove, id})

  def init(_) do
    tick()

    # Here we use a MapSet as it will reject duplicate items.
    # If you wish to store more data with a task than just a user ID,
    # use a Map instead.
    {:ok, MapSet.new}
  end

  defp tick, do: Process.send_after(self(), :tick, @tick_interval)

  def handle_cast({:add, id}, tasks), do: {:noreply, MapSet.put(tasks, id)}

  def handle_cast({:remove, id}, tasks), do: {:noreply, MapSet.delete(tasks, id)}

  def handle_info(:tick, tasks) do
    tasks = process_tasks(tasks)

    tick()

    {:noreply, tasks}
  end

  defp process_tasks(tasks) do
    tasks_list =
      Enum.filter tasks, fn(id) ->
        # Here we process the individual task, such as polling
        # the external API.
        case ExternalAPI.poll(id) do
          # We return true from this anonymous function if the task should
          # stay, and false if the task should be removed.
          "terminate = true" -> true
          "terminate = false" -> false
        end
      end

    # `Enum.filter/2` returns a list, so we must convert it back into a MapSet.
    MapSet.new(tasks_list)
  end

  def handle_info(_, tasks), do: {:noreply, tasks}
end

Thanks.

In you example In process_tasks a list of tasks is iterated over and an http request ExternalAPI.poll is sent syncronously and therefore waited for a response syncronously as well. Is that a wise thing to do? Why not create for each http request a separate thread?

You’re right, that is better. Instead of Enum.filter/2 you might consider just using Enum.each/2 and calling MyWorker.remove/1 from the process handling the individual task if the task should be removed.

Should that be done via spawn? something like this:

spawn(Helloer, :my_func, [123, "something"])

Then how would I do filtering if all them are in different thread? Will I have to wait something for all them for finish in each iteration?

I would use a Task.

As I mentioned, if you go that route you may not want to use Enum.filter/2. Instead, call the remove/1 function from inside the Task process with the user ID of the task to be removed.

Enum.each tasks, fn(id) ->
  Task.start fn ->
    if ExternalAPI.poll(id) == "terminate = true", do: remove(id)
  end
end

Also, how is an instance of MapSet stored between http requests? I mean, an http request to an html page: a user reloads a page with a button and … a list of tasks is gone? In a session or is there more idiomatic way?

I’m not entirely sure what you’re asking.

The MapSet is stored as the MyWorker GenServer’s state on the server. The MyWorker GenServer is a continuously running process that should be started as part of your application’s supervision tree when your application starts. You should not be starting and stopping it on demand when a user loads or reloads a page.

There’s no variable in MyWorker which stores MapSet. Where is it stored?

Will you show me a simple example of a client of MyWorker?

The MyWorker uses the :name option as part of the start_link call, which means that the process is registered under a global (node-wide) name. The other calls that use GenServer.call and GenServer.cast use the same name to access it later.

So this means that the process will remain available after starting until your application closes, the process crashes, or you call GenServer.stop/2 from somewhere. In any case, the process will remain regardless of the connecting/disconnecting of users to your web-facing interface.

that’s not what I asked about.