Elixir Project: Event Bus
An event bus is a GenServer that passes events from publishers to subscribers by topic. A subscriber only receives events for the topics it registered for.
A subscriber receives events in order
One process subscribes to a topic and collects what arrives. Its log shows the events in publish order.
defmodule Bus do
use GenServer
def start_link(_arg), do: GenServer.start_link(__MODULE__, %{subs: %{}})
def init(state), do: {:ok, state}
def handle_call({:subscribe, topic}, {pid, _tag}, state) do
current = Map.get(state.subs, topic, [])
subs = if pid in current, do: state.subs, else: Map.put(state.subs, topic, current ++ [pid])
{:reply, :ok, %{state | subs: subs}}
end
def handle_call({:unsubscribe, topic}, {pid, _tag}, state) do
current = Map.get(state.subs, topic, [])
subs = Map.put(state.subs, topic, List.delete(current, pid))
{:reply, :ok, %{state | subs: subs}}
end
def handle_call({:publish, topic, event}, _from, state) do
for pid <- Map.get(state.subs, topic, []) do
send(pid, {:event, topic, event})
end
{:reply, :ok, state}
end
end
{:ok, bus} = Bus.start_link([])
parent = self()
listener = spawn(fn ->
collect = fn collect, acc ->
receive do
{:event, topic, text} -> collect.(collect, acc ++ [{:event, topic, text}])
{:report, from} ->
send(from, {:log, acc})
collect.(collect, acc)
:stop -> :ok
end
end
GenServer.call(bus, {:subscribe, :news})
send(parent, :ready)
collect.(collect, [])
end)
receive do :ready -> :ok end
GenServer.call(bus, {:publish, :news, "Win"})
GenServer.call(bus, {:publish, :news, "Draw"})
send(listener, {:report, self()})
receive do {:log, log} -> IO.inspect(log) end
send(listener, :stop)
GenServer.stop(bus)Other topics never arrive
The subscriber registered for news only. An event on another topic is not delivered to it.
defmodule Bus do
use GenServer
def start_link(_arg), do: GenServer.start_link(__MODULE__, %{subs: %{}})
def init(state), do: {:ok, state}
def handle_call({:subscribe, topic}, {pid, _tag}, state) do
current = Map.get(state.subs, topic, [])
subs = if pid in current, do: state.subs, else: Map.put(state.subs, topic, current ++ [pid])
{:reply, :ok, %{state | subs: subs}}
end
def handle_call({:unsubscribe, topic}, {pid, _tag}, state) do
current = Map.get(state.subs, topic, [])
subs = Map.put(state.subs, topic, List.delete(current, pid))
{:reply, :ok, %{state | subs: subs}}
end
def handle_call({:publish, topic, event}, _from, state) do
for pid <- Map.get(state.subs, topic, []) do
send(pid, {:event, topic, event})
end
{:reply, :ok, state}
end
end
{:ok, bus} = Bus.start_link([])
parent = self()
listener = spawn(fn ->
collect = fn collect, acc ->
receive do
{:event, topic, text} -> collect.(collect, acc ++ [{:event, topic, text}])
{:report, from} ->
send(from, {:log, acc})
collect.(collect, acc)
:stop -> :ok
end
end
GenServer.call(bus, {:subscribe, :news})
send(parent, :ready)
collect.(collect, [])
end)
receive do :ready -> :ok end
GenServer.call(bus, {:publish, :sports, "Goal"})
GenServer.call(bus, {:publish, :news, "Win"})
send(listener, {:report, self()})
receive do {:log, log} -> IO.inspect(log) end
send(listener, :stop)
GenServer.stop(bus)Unsubscribing stops delivery
The script itself subscribes, receives one event, then leaves the topic. The next event on that topic never arrives.
defmodule Bus do
use GenServer
def start_link(_arg), do: GenServer.start_link(__MODULE__, %{subs: %{}})
def init(state), do: {:ok, state}
def handle_call({:subscribe, topic}, {pid, _tag}, state) do
current = Map.get(state.subs, topic, [])
subs = if pid in current, do: state.subs, else: Map.put(state.subs, topic, current ++ [pid])
{:reply, :ok, %{state | subs: subs}}
end
def handle_call({:unsubscribe, topic}, {pid, _tag}, state) do
current = Map.get(state.subs, topic, [])
subs = Map.put(state.subs, topic, List.delete(current, pid))
{:reply, :ok, %{state | subs: subs}}
end
def handle_call({:publish, topic, event}, _from, state) do
for pid <- Map.get(state.subs, topic, []) do
send(pid, {:event, topic, event})
end
{:reply, :ok, state}
end
end
{:ok, bus} = Bus.start_link([])
GenServer.call(bus, {:subscribe, :news})
GenServer.call(bus, {:publish, :news, "Win"})
IO.inspect(receive do {:event, topic, text} -> {topic, text} end)
GenServer.call(bus, {:unsubscribe, :news})
GenServer.call(bus, {:publish, :news, "Later"})
IO.inspect(receive do {:event, _topic, _text} -> "got one" after 0 -> "none" end)
GenServer.stop(bus)Example
defmodule Bus do
use GenServer
def start_link(_arg), do: GenServer.start_link(__MODULE__, %{subs: %{}})
def init(state), do: {:ok, state}
def handle_call({:subscribe, topic}, {pid, _tag}, state) do
current = Map.get(state.subs, topic, [])
subs = if pid in current, do: state.subs, else: Map.put(state.subs, topic, current ++ [pid])
{:reply, :ok, %{state | subs: subs}}
end
def handle_call({:unsubscribe, topic}, {pid, _tag}, state) do
current = Map.get(state.subs, topic, [])
subs = Map.put(state.subs, topic, List.delete(current, pid))
{:reply, :ok, %{state | subs: subs}}
end
def handle_call({:publish, topic, event}, _from, state) do
for pid <- Map.get(state.subs, topic, []) do
send(pid, {:event, topic, event})
end
{:reply, :ok, state}
end
end
{:ok, bus} = Bus.start_link([])
parent = self()
listener = spawn(fn ->
collect = fn collect, acc ->
receive do
{:event, topic, text} -> collect.(collect, acc ++ [{:event, topic, text}])
{:report, from} ->
send(from, {:log, acc})
collect.(collect, acc)
{:unsubscribe, topic, from} ->
GenServer.call(bus, {:unsubscribe, topic})
send(from, :unsubscribed)
collect.(collect, acc)
:stop -> :ok
end
end
GenServer.call(bus, {:subscribe, :news})
GenServer.call(bus, {:subscribe, :sports})
send(parent, :ready)
collect.(collect, [])
end)
receive do :ready -> :ok end
GenServer.call(bus, {:publish, :news, "Win"})
GenServer.call(bus, {:publish, :sports, "Goal"})
send(listener, {:unsubscribe, :sports, self()})
receive do :unsubscribed -> :ok end
GenServer.call(bus, {:publish, :sports, "Penalty"})
GenServer.call(bus, {:publish, :news, "Draw"})
send(listener, {:report, self()})
receive do {:log, log} -> IO.inspect(log) end
send(listener, :stop)
GenServer.stop(bus)Key points
- Subscribe: a subscriber registers its interest in a topic.
- Publish: publishing sends an event to every subscriber of that topic.
- Order: events reach a subscriber in the order they were published.
- Unsubscribe: a subscriber that leaves a topic stops receiving its events.
- Topics: events on other topics never reach a subscriber.
In this example, one process subscribes to news and sports, then receives two events in publish order. It leaves the sports topic, so the later sports event is not delivered, while the later news event still arrives. The listener is stopped and the bus is stopped at the end.
Try it Yourself
Edit the code, then press Run.
Running is simulated
Exercise
A process subscribes to news only. The bus publishes a sports event and then a news event. What does the process receive?