Qqwy

Qqwy

TypeCheck Core Team

How to wrap Phoenix PubSub?

I am working on a small library that computes a moving average, which would work well with Phoenix PubSub, in that it could subscribe itself to a topic using Phoenix PubSub, and then handle the events as they come in.

However, users should be able to modify the exact logic that is going on inside, including what topics it is subscribed to. This is a bit of a problem though, because a PubSub-broadcasted message does not contain any identifying information (It is just a ‘plain’ message that will trigger your GenServer’s handleinfo.)

So, if I want to give users the ability to subscribe to varying PubSub topics, they need to be able to tap into the handleinfo part of the GenServer that the library constructs, to allow matching on the proper messages.

What would be a proper way to do this?

Most Liked

chrismccord

chrismccord

Creator of Phoenix

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?

windholtz

windholtz

Qqwy

Qqwy

TypeCheck Core Team

Thank you very much for your help, everyone. :heart:

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.

Where Next?

Popular in Questions Top

vegabook
I’m brand new to Phoenix and I have stripped one of the demo applications to the bone. I just want to get an svg up on the screen. Here i...
New
vonH
In asking this question I am more interested about the expressiveness of the language itself and less concerned about the availability of...
New
belgoros
I’m not a pro in using Regex and can’t figure out why the following behaviour happens, especially if we take into account the difference ...
New
siddhant3030
Hi, I have to write a raw query for one of my project. But till now I have used ecto queries and don’t have much experience writing raw ...
New
WestKeys
Currently suffering from paralysis by [HTTP client] analysis. This is rather unusual in Elixirland as there tends to be consensus on the ...
New
fayddelight
I tried installing elixir 1.11.2 erlang 23.3.4 via asdf in my zsh shell. Enabled the versions locally and globally. When I list them ...
New
svb
Hi! Currently I want to submit a form by pressing the Enter key. However, since my input field is of type “textarea” this is just adds a...
New

Other popular topics Top

nobody
Hi! In PHP: $_SERVER[‘SERVER_ADDR’] - in Elixir? Searched the docs for ip address and the web, no good results. Thanks!
New
JeremM34
Hello, how can I check the Phoenix version ? Thanks !
New
sen
Hi All, I set a environment variables in dev.exs , like below code. when i start server, how can i set the ${enable} value? thanks. d...
New
sergio_101
I am VERY much an elixir newbie. I have taken one elixir course and one phoenix course on Udemy. During that course, I saw the instructor...
New
Patoshizzle
After calling mix ecto.create I get this error: 17:00:32.162 [error] GenServer #PID&lt;0.412.0&gt; terminating ** (Postgrex.Error) FATAL...
New
TunkShif
This post is an instruction guide to help you setup your Neovim for Elixir development from scratch. It includes general information on h...
274 42533 114
New

We're in Beta

About us Mission Statement