I still don’t have a complete picture, especially around users making their own events, but let’s start wit this code. Here we have a GameStore which is the boundary for the game system. You are correct that you want to wrap the pubsub mechanism so the caller doesn’t need to know about it. Given different actions in the system, like purchasing or transferring an item, you’ll broadcast events as part of the success cases. Meanwhile, you can expose a subscribe/0, subscribe/1 and unsubscribe functions on the GameStore for other callers to consume. If they are interested in stats they can subscribe to the “stats” group. I show the Caller example below the GameStore:

defmodule GameStore do

  def subscribe do
    Phoenix.PubSub.subscribe(GameStore.PubSub, "all")
  end

  def subscribe(group) do
    Phoenix.PubSub.subscribe(GameStore.PubSub, group)
  end

  def transfer_item(%Item{owner_id: owner_id} = item, %Player{} = purchaser) do
    case transfer_ownership(purchaser) do
      {:ok, item, %Transfer{} = tranfer} -> 
        broadcast({:transfers, :completed, transfer})
        {:ok, item, transfer}
      {:error , ...} -> ...
    end
  end

  defp broadcast({group, event, term}) do
    Phoenix.PubSub.broadcast(GameStore.PubSub, group, {group, event, term})
    Phoenix.PubSub.broadcast(GameStore.PubSub, "all", {group, event, term})
  end
end


defmodule SomeServer do

  def init(opts) do
    :ok = GameStore.subscribe(:transfers)
    :ok = GameStore.subscribe(:stats)
    ...
    {:ok, some_state}
  end

  def handle_info({:stats, :moving_avarage, ave}, state) do
    IO.puts "the new moving average of the store is #{inspet ave}"
    {:noreply, state}
  end

  def handle_info({:transfers, :completed, trans}, state) do
    IO.puts "A transfer was completed from" <>
            "#{trans.from_player_id} to #{trans.to_player_id}"
    {:noreply, state}
  end
end

Does this put you on the right track?

updated code available here:

Thank you very much for your help, everyone.

To make it more clear what I am struggling with, I have written it down now as a library: SlidingWindow. (Still very work-in-progress; it does not have a lot of documentation yet as things might still change a lot.)

It does what I want to perform in a finite-state-automaton kind of way; that is, SlidingWindows does not care about GenServers or anything at all right now, but only about gathering and updating the information stored in the struct.

An example of a behaviour to calculate averages: (See also the file test_helper.exs)

defmodule TestSW do
  @behaviour SlidingWindow.Behaviour

  defmodule Transaction do
    defstruct [:value, :created_at]
    def new(value, created_at) do
      %__MODULE__{value: value, created_at: created_at}
    end
  end

  defstruct count: 0, sum: 0, product: 1

  def empty_aggregate() do
    %__MODULE__{}
  end

  def add_item(agg, %Transaction{value: int}) do
    %__MODULE__{agg |
      count: agg.count + 1,
      sum: agg.sum + int,
      product: agg.product * int
    }
  end

  def remove_item(agg, %Transaction{value: int}) do
    %__MODULE__{agg |
      count: agg.count - 1,
      sum: agg.sum - int,
      product: div(agg.product, int)
    }
  end

  def extract_timestamp(%Transaction{created_at: timestamp}) do
    timestamp
  end

end

And then you can create it using something like:

result = 
      SlidingWindow.init(TestSW, 10, Timex.Duration.from_seconds(2), initial_data())
      |> SlidingWindow.shift_stale_items(ten_sec_future)
      |> SlidingWindow.add_item(%Transaction.new(10, Timex.now())
      |> SlidingWindow.add_item(%Transaction.new(15, Timex.now())
      # And at some point:
      |> SlidingWindow.get_aggregates()

# Result now contains something like:
result ==            %{0 => %TestSW{count: 0, product: 1, sum: 0},
                       1 => %TestSW{count: 0, product: 1, sum: 0},
                       2 => %TestSW{count: 0, product: 1, sum: 0},
                       3 => %TestSW{count: 1, product: 1, sum: 1},
                       4 => %TestSW{count: 2, product: 6, sum: 5},
                       5 => %TestSW{count: 2, product: 20, sum: 9},
                       6 => %TestSW{count: 2, product: 42, sum: 13},
                       7 => %TestSW{count: 2, product: 72, sum: 17},
                       8 => %TestSW{count: 2, product: 110, sum: 21},
                       9 => %TestSW{count: 2, product: 156, sum: 25}}

In the end, I think that the smartest decision I can make is to not mess with wrapping this inside of a GenServer-like layer myself, as there are an infinite amount of possible ways how someone might want to manipulate the FSA. Rather, I’ll write some example code on how it could be wrapped in your own GenServer including an send_after message to ensure that the data inside never becomes stale.

I don’t have a full picture of your use case form your description, can you describe a little more what you mean by:

Maybe showing a pseudo client/server api that you’re thinking of would help. What I can say is if you need certain info to be a part of the message, you can make your own message contract that includes this info, just like the Phoenix.SocketBroadcast for channels contains extra info, your contract could be

{group, event, term}, for example:

{:stats, :new_average, %{value: 100}}
{:health, :cpu_spike, %{node: node(), load: ...}}

Make sense?

They are basically no different from normal messages, so that then begs the question: How do you handle arbitrary messages sent to your process (any process, pubsub or not)?

Also I’m curious what you mean by this? An info message is just a non-genserver-wrapped message, very very common in GenServers?