by Chris Erin on July 25, 2019

I like to dream about robots. That's a pretty broad statement so let me narrow that down. I like to dream about semi-automonous 4 wheeled rovers, moving around the surface of the moon controlled by teams of scientists, students, and hobbyists via a command line interface. OK, that's a bit more specific.

Alas, it's just a dream, but it gets me imagining different ways that Elixir and Processes might work to facilitate the robot rover's operation.

This post is about an imagined light sensing sub system. Light is sensed with a sensor and when that sensor changes it sends a message to a broker which then forwards the message to it's subscribers. I want to test this system with ExUnit, both to get a feel for how the different parts fit together and to get a feel for how to test GenServers. For my purposes this system translates into 3 GenServers:

• LightSensor - a GenServer that interfaces with a hardware sensor and sends messages to a broker.
• MessageBroker - a GenServer that manages subscribed processes and sends messages to those processes
• LightServer - a GenServer that is subscribed to light messages, receives light messages and makes decisions based on those messages

In this post I will go through the code of each GenServer to show how each is implemented, how they work together and how they can be tested.

This GenServer would provide the interface for the hardware. I'm not really a hardware person (despite having attended the Nerves Workshop!). While I can imagine a GenServer being the interface to a sensor like this, I don't particularly care about how that's implemented.

What I do care about is that the GenServer produces a steady stream of messages that might look like what a light sensor might produce. Because I plan to test drive this with ExUnit, instead of actual hardware, I want a MockLightServer that will produce data that looks and behaves like real data.

That will look something like this:

defmodule LightSensor.Data do
  def data do
    1..100
    |> Enum.map(fn n ->
      {Enum.random(0..n), Enum.random(0..100)}
    end)
  end
end

defmodule MockLightSensor do
  use GenServer

  @name __MODULE__

  def start_link(message_broker_pid, start_end_notifier_pid) do
    result =
      {:ok, _pid} =
      GenServer.start_link(
        __MODULE__,
        %{
          message_broker: message_broker_pid,
          start_end_notifier: start_end_notifier_pid,
          data: LightSensor.Data.data()
        },
        name: @name
      )

    result
  end

  def start_messages() do
    GenServer.cast(@name, :next)
  end

  @impl true
  def init(args) do
    {:ok, args}
  end

  @impl true
  def handle_info(:next, state) do
    GenServer.cast(@name, :next)
    {:noreply, state}
  end

  @impl true
  def handle_cast(:next, %{
        message_broker: broker,
        start_end_notifier: start_end_notifier_pid,
        data: [{next_light_message, _} | []]
      }) do
    GenServer.cast(broker, {:light_message, next_light_message})
    # signal to the test process that the data sending has ended
    send(start_end_notifier_pid, :ended)
    {:noreply, %{}}
  end

  @impl true
  def handle_cast(:next, %{
        message_broker: broker,
        start_end_notifier: start_end_notifier_pid,
        data: [{interval, next_light_message} | rest]
      }) do

    Process.send_after(self(), :next, interval)
    GenServer.cast(broker, {:light_message, next_light_message})
    {:noreply, %{message_broker: broker, start_end_notifier: start_end_notifier_pid, data: rest}}
  end
end

This process has three keys in it's state map, message_broker, start_end_notifier and data.

• message_broker - the pid that it will send light messages to
• start_end_notifier - the pid that will be notified when there is no more data
• data - the mock data in the format of {interval, next_light_message}

This GenServer is sending messages to three different places. The first is imitating how the light sensor might behave by sending light messages to the message broker. The second and third messages are for testing purposes. The test needs to know when it's done, so the GenServer will send a message to the test process when it's out of data. The GenServer needs to know when to send another message, so it will take the interval message from the data tuple and send a message to itself after the interval has passed.

This GenServer follows a pub/sub pattern. A pub/sub GenServer here will allow many different business logic processes to subscribe to light messages without the LightSensor server needing to know or manage processes that might subscribe temporarily.

It looks like this:

defmodule MessageBroker do
  use GenServer

  @name __MODULE__

  def start_link() do
    GenServer.start_link(@name, %{subscriptions: []})
  end

  @impl true
  def init(args) do
    {:ok, args}
  end

  @impl true
  def handle_cast({:subscribe, pid}, %{subscriptions: subs}) do
    {:noreply, %{subscriptions: [pid | subs]}}
  end

  @impl true
  def handle_cast({:light_message, message}, %{subscriptions: subs}) do
    Enum.each(subs, fn sub_pid ->
      GenServer.cast(sub_pid, {:light, message})
    end)

    {:noreply, %{subscriptions: subs}}
  end
end

This version only handles two messages subscribe and light_message. subscribe messages come from interested processes and light_message messages come from the LightSensor itself.

When receiving a LightSensor message it will loop through all the subscribed processes and send a light message to each.

The LightServer will attach meaning and behavior to the messages it receives. This is where the business logic would live. Conceivably, the LightServer would make decisions about where to move, how much energy to use, whom to notify, or maybe even when to fold or unfold the solar panels. In this example, it's pretty dumb, just recording the light messages that are above a certain value.

defmodule LightServer do
  use GenServer
  @name __MODULE__

  @impl true
  def init(args) do
    {:ok, args}
  end

  def start_link(broker) do
    {:ok, pid} = GenServer.start_link(__MODULE__, %{very_bright_measurements: []}, name: @name)
    GenServer.cast(broker, {:subscribe, pid})
    {:ok, pid}
  end

  def get_very_hot_temperatures do
    GenServer.call(@name, :get_very_hot_temperatures)
  end

  @impl true
  def handle_cast({:light, light_amount}, %{very_bright_measurements: amounts}) when light_amount > 90 do
    {:noreply, %{very_bright_measurements: [light_amount | amounts]}}
  end

  @impl true
  def handle_cast({:light, _light_amount}, state) do
    # do nothing
    {:noreply, state}
  end

  @impl true
  def handle_call(:get_very_bright_measurements, _from, %{very_bright_measurements: amounts} = state) do
    {:reply, amounts, state}
  end
end

What makes programming this subsystem fun is being able to test it. I want to test that the LightServer has data when all the processes have been sent.

The ExUnit test will look like this:

defmodule LightSensorTest do
  use ExUnit.Case

  setup do
    {:ok, message_broker_pid} = MessageBroker.start_link()
    {:ok, _pid} = MockLightSensor.start_link(message_broker_pid, self())
    {:ok, _pid} = LightServer.start_link(message_broker_pid)
  end

  test "Does Light Server receive messages?" do
    MockLightSensor.start_messages()
    assert_receive(:ended, 10000)
    assert very_bright_measurements = LightServer.get_very_bright_measurements()
    assert(length(very_bright_measurements) > 0)
  end
end

It's a simple test, just assert that the LightServer has received the expected data. This assertion will fail, however, without allowing all the messages to fly around and be processed. Because the messages are almost all cast (read: asynchronous) the test would reach it's end before anything had happened without the assert_received call. This function waits for a specific message to be sent to the test process in a certain amount of time. In this case, we wait 10 seconds for all the messages to be processed and for the MockLightSensor to send the :ended message to the test process.

This imagined subsystem isn't imagined exhaustively. It's just a sketch of something that might exist in a future rover robot. It's an exercise that I can use to get better at understanding Erlang/Elixir process patterns and to get better at testing processes in general. In the future it might be fun to imagine other rover robot subsystems.

Image via NASA