Home Posts Post Search Tag Search

Real World Event Sourcing - 04 - Injectors and Notifiers
Published on: 2026-06-29 Tags: elixir, Blog, State, Systems, Supervisor Tree, event-sourcing, Real World Event Sourcing

3. Enforcing Perimeters with Injectors and Notifiers

Injector and Notifier. This will help us to build a way to notify and inject events into a system.



Handling Input and Output in an Event-Sourced World

We need a way to allow for the “real world” to access the information that we get throughout the life-cycle of the series of events. An event can only have a single value for it, and as such we are not as privy to the real current state of the data. We then need a way to react to injected events.



Reacting to Injected Events

It’s easy to just react to an event but the real catch is what and when to react, and even what to inject. There might be a call for not even injecting the event or inject but not Notify, here are some questions to ask:

• Does the external event represent something important to the components
in your system?
• Does the event occur without the aid of an internal command?
• Does the occurrence of the event have meaning beyond the ephemeral
state?
• Does the occurrence of the event also affect the state of an aggregate in
your system?

Something like the current state of the light-bulb might only need a notifier and something like a motion sensor attached to a camera might want to be injected an notified.



Notifying External Consumers

This is where we take those events and notify the outside world. One of the most common is an event listener that will look for failures and then notify a Slack or other messaging service. Here is a few common examples:

• Submit a label request to a third-party shipper when an order has been placed
• Email a customer when the status of an order changes
• Publish federated chat messages from internal chats to an external source
• Invoke webhooks in response to certain events

These are things that need to be sent out as a notification, you would also think about injecting them into the system as well. One thing to mention is that the Notify is not the code to further trigger events or logging, that would be the job of the process manager, keep these tasks separate.



Introducing Cloud Events

No matter what we are doing there will be a payload that needs to be sent with every notification as well as the payload for the injector. We should then try to use things like CRDT’s (conflict-free replicated data types). This will help us to allow more than one system to use the same data type and not have to many reducers. Cloud Events will help us with that here. Once you have a standard way of sending information you will be able to have systems use the same information.


Here are some of the needed or used data within a cloud event:

• specversion—The version of Cloud Events to which this event conforms
• type—A fully qualified name of the event type, for example, org.thisbook.that-
sample.this_event
• source—A free-form identifier indicating the originator of the event
• id—A unique identifier for the event. See the specification for rules
regarding what constitutes a valid ID
• datacontenttype—A mime type that indicates the type of data that can be
found in the data field
• time—An ISO 8601 string formatted timestamp
• data—The actual payload

For the rest of the book if there is a format that isn’t specified we will be using Cloud Events JSON. Here is a small snippet of code that would allow you to create a Cloud Event JSON.

defp new_cloudevent(type, data) do
  %{
    "specversion" => "1.0",
    "type" => "org.book.flighttracker.#{String.downcase(type)}",
    "source" => "radio_aggregator",
    "id" => UUID.uuid4(),
    "datacontenttype" => "application/json",
    "time" => DateTime.utc_now() |> DateTime.to_iso8601(),
    "data" => data
  }
  |> Cloudevents.from_map!()
  |> Cloudevents.to_json()
end

## This will produce something like
{
  "specversion" : "1.0",
  "type" : "org.book.flighttracker.aircraft_identified",
  "source" : "radio_aggregator",
  "subject": null,
  "id" : "0f749a4f-6f2c-4301-9521-21a9459080b4",
  "time" : "2024-04-05T17:31:00Z",
  "data" : { ... }
}


Building a Flight Tracker with Injection and Notification

For this next set of code we want to build a flight tracker that will take event data from a system and then allow a user to subscribe to the data stream for that flight. It will involve setting up a GenStage and then inputting information into the server. I’ll go over the code as we build it.


For this we will use the ADS-B (_Automatic Dependent Surveillance-Broadcast). This is a standard for logging flights. We will take these events and inject them into the internal event stream then we will allow a party to get information about the flight that they want to know about. We will need to implement these events:

• Aircraft Identified
• Squawk Received
• Position Reported
• Velocity Reported

First let’s create the new project and then build the MessageBroadcaster.

mix new flight_tracker --sup
defmodule FlightTracker.MessageBroadcaster do
  use GenStage
  require Logger

  def start_link(_) do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  @doc """
  Injects a raw message that is not in cloud event format
  """
  def broadcast_message(message) do
    GenStage.call(__MODULE__, {:notify, message})
  end

  @doc """
  Injects a cloud event to be published to the stage pipeline
  """
  def broadcast_event(evt) do
    GenStage.call(__MODULE__, {:notify_evt, evt})
  end

  @impl true
  def init(:ok) do
    {:producer, :ok, dispatcher: GenStage.BroadcastDispatcher}
  end

  @impl true
  def handle_call({:notify, message}, _from, state) do
    {:reply, :ok, [to_event(message)], state}
  end

  @impl true
  def handle_call({:notify_evt, evt}, _from, state) do
    {:reply, :ok, [evt], state}
  end

  @impl true
  def handle_demand(_demand, state) do
    {:noreply, [], state}
  end

  defp to_event(%{
         type: :aircraft_identified,
         message: %{icao_address: _icao, callsign: _callsign, emitter_category: _cat} = msg
       }) do
    new_cloudevent("aircraft_identified", msg)
  end

  defp to_event(%{type: :squawk_received, message: %{squawk: _squawk, icao_address: _icao} = msg}) do
    new_cloudevent("squawk_received", msg)
  end

  defp to_event(%{
         type: :position_reported,
         message: %{
           icao_address: icao,
           position: %{
             altitude: alt,
             longitude: long,
             latitude: lat
           }
         }
       }) do
    new_cloudevent("position_reported", %{
      altitude: alt,
      longitude: long,
      latitude: lat,
      icao_address: icao
    })
  end

  defp to_event(%{
         type: :velocity_reported,
         message:
           %{heading: _head, ground_speed: _gs, vertical_rate: _vr, vertical_rate_source: vrs} =
             msg
       }) do
    source =
      case vrs do
        :barometric_pressure -> "barometric"
        :geometric -> "geometric"
        _ -> "unknown"
      end

    new_cloudevent("velocity_reported", %{msg | vertical_rate_source: source})
  end

  defp to_event(msg) do
    Logger.error("Unknown message: #{inspect(msg)}")
    %{}
  end

  defp new_cloudevent(type, data) do
    %{
      "specversion" => "1.0",
      "type" => "org.book.flighttracker.#{String.downcase(type)}",
      "source" => "radio_aggregator",
      "id" => UUID.uuid4(),
      "datacontenttype" => "application/json",
      "time" => DateTime.utc_now() |> DateTime.to_iso8601(),
      "data" => data
    }
    |> Cloudevents.from_map!()
    |> Cloudevents.to_json()
  end
end

First GenStage is better suited for this as its made to deal with a pipeline and it is made to make sure that a injector will need to ask for the information that it needs and no more. This solves a few things:

* It keeps track of a users requests and will fill them when the information 
comes in.
* helps to ensure that there will be no real backlog of events seen but not processed.
* This is a pull based event handler, so it will only give what it has and not any more.
* This might make is slower for processing to many events but it will ensure that the injector and the notifier will run as we want them to.

We also use an ETS table to ensure easy access to the information and ensure that we have unique ID’s for the entries.


Getting ADS-B Messages

Okay so now that we have a way of getting the messages into the system we need a way to read messages. There is a file within the course source code that has quite a few of the flights, let’s add that to the project and then build a way of reading the files. First take the file and add it to the top folder of the project then create this file_injector.ex

defmodule FlightTracker.FileInjector do
  alias FlightTracker.MessageBroadcaster
  use GenServer
  require Logger

  def start_link(file) do
    GenServer.start_link(__MODULE__, file, name: __MODULE__)
  end

  @impl true
  def init(file) do
    Process.send_after(self(), :read_file, 2_000)
    {:ok, file}
  end

  @impl true
  def handle_info(:read_file, file) do
    File.stream!(file)
    |> Enum.map(&String.trim/1)
    |> Enum.each(fn evt -> MessageBroadcaster.broadcast_event(evt) end)

    {:noreply, file}
  end
end

Next start up an iex session.

iex> GenServer.start_link(FlightTracker.FlightNotifier, "PRAG250"

Not we can create the notifier for the events

defmodule FlightTracker.FlightNotifier do
  alias FlightTracker.MessageBroadcaster
  alias FlightTracker.CraftProjector
  require Logger
  use GenStage

  def start_link(flight_callsign) do
    GenStage.start_link(__MODULE__, flight_callsign)
  end

  def init(callsign) do
    {:consumer, callsign, subscribe_to: [MessageBroadcaster]}
  end

  # GenStage callback for consumers
  def handle_events(events, _from, state) do
    for event <- events do
      handle_event(Cloudevents.from_json!(event), state)
    end

    {:noreply, [], state}
  end

  defp handle_event(
         %Cloudevents.Format.V_1_0.Event{
           type: "org.book.flighttracker.position_reported",
           data: dt
         },
         callsign
       ) do
    aircraft = CraftProjector.get_state_by_icao(dt["icao_address"])
    # it's possible that we don't have the callsign yet
    if String.trim(Map.get(aircraft, :callsign, "")) == callsign do
      Logger.info("#{aircraft.callsign}'s position: #{dt["latitude"]}, #{dt["longitude"]}")
    end
  end

  defp handle_event(_evt, _state) do
    # we're not interested in anything else
  end
end

Running the Flight Tracker

Now that we have the code for dealing with reading and logging the flights we need to set up the application.ex so that we start all the servers when we run the application.

defmodule FlightTracker.Application do
  use Application
  @impl true
  def start(_type, _args) do
    children = [
      {FlightTracker.FileInjector, ["./sample_cloudevents.json"]},
      {FlightTracker.MessageBroadcaster, []},
      {FlightTracker.CraftProjector, []},
      {FlightTracker.FlightNotifier, "AMC421"}
    ]

    opts = [strategy: :rest_for_one, name: FlightTracker.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

The children automatically started are the following:

• The file injector (which starts injecting after a 2s delay to give humans a
chance to see things start up)
• The message broadcaster (a GenStage producer)
• The craft projector (a GenStage consumer)
• A sample flight notifier preconfigured to a certain flight (another GenStage
consumer)

Lastly we need to deal with the mix.ex inode to make sure that we have the right dependencies.

defmodule FlightTracker.MixProject do
  use Mix.Project

  def project do
    [
      app: :flight_tracker,
      version: "0.1.0",
      elixir: "~> 1.17",
      start_permanent: Mix.env() == :prod,
      deps: deps()
    ]
  end

  # Run "mix help compile.app" to learn about applications.
  def application do
    [
      extra_applications: [:logger],
      mod: {FlightTracker.Application, []}
    ]
  end

  # Run "mix help deps" to learn about dependencies.
  defp deps do
    [
      {:gen_stage, "~> 1.2.1"},
      {:cloudevents, "~> 0.6.1"},
      {:uuid, "~> 1.1"}
    ]
  end
end

Now we can run the program with iex -S mix



Wrapping Up

This chapter as well as getting us a fun program to use injectors and notifiers allowed us to better understand when and why we would use either method. Keep in mind that not all events need any of the 2, while some will need both. Make sure to talk to your team about the needs of the user and the company.

Back to posts