I enjoy IoT development using Elixir programming language, Nerves IoT platform and Phoenix web framework. It is so much fun. I was able to build a real-time temperature and humidity monitoring system for my living room. My API server accepts a sensor measurement then broadcasts the update to all the users of my real-time dashboard.

Today I am going to write about real-time page refresh using Phoenix LiveView and Phoenix PubSub so that I can quickly implement the same thing in the future.

4/3(土) 00:00〜 4/5(月) 23:59開催のautoracex #21での成果です。

日本語版

Implement PubSub utilities in a context module

First of all, I prepare utility functions for subscribing and broadcasting messages in a given context module. Here I add subscribe/0 and broadcast/2 to the Example.Environment context module. I use inspect(__MODULE__) as a topic so that I can ensure that the topic name is unique as well as saving time for decision making on the topic name and discovery of the topic name.

 defmodule Example.Environment do

   ...
+
+  @topic inspect(__MODULE__)
+
+  @doc """
+  Subscribe to this context module's messages.
+  """
+  def subscribe do
+    Phoenix.PubSub.subscribe(Example.PubSub, @topic)
+  end
+
+  @doc """
+  Broadcast a message to the subscribers when something happens.
+  """
+  def broadcast({:ok, record}, event) do
+    Phoenix.PubSub.broadcast(Example.PubSub, @topic, {event, record})
+    {:ok, record}
+  end
+
+  def broadcast({:error, _} = error, _event), do: error
+

Broadcast a message as needed

Now that I have PubSub utility functions, I can broadcast a message when something happens. In the following example, I notify all the subscribers of a newly-inserted measurement record.

 defmodule Example.Environment do

   ...

   @doc """
   Creates a measurement.

   ## Examples

       iex> create_measurement(%{field: value})
       {:ok, %Measurement{}}

       iex> create_measurement(%{field: bad_value})
       {:error, %Ecto.Changeset{}}

   """
   def create_measurement(attrs \\ %{}) do
     %Measurement{}
     |> Measurement.changeset(attrs)
     |> Repo.insert()
+    |> broadcast(:measurement_inserted)
   end

Subscribe to PubSub topic when LiveView is connected

In a LiveView, I subscribe to a PubSub topic using the PubSub utilities prepared above. It is important that the subscription has to be done after the LiveView connecton has been established.

 defmodule ExampleWeb.EnvironmentLive do
   use ExampleWeb, :live_view

   ...

   @impl true
   def mount(_params, _session, socket) do
+    if connected?(socket) do
+      Environment.subscribe()
+    end

     ...

     {:ok, socket, temporary_assigns: [measurements: []]}
   end

Handle PubSub message as needed

Once a LiveView has subscribed to a PubSub topic, it will receive a message whenever a mesage is broadcast on the subscribed topic. I can handle the event, pattern-matching the event in handle_info/2. In this example, an event is a tuple like {event_name, new_record} because that is the format that my broadcast function uses.

 defmodule ExampleWeb.EnvironmentLive do
   use ExampleWeb, :live_view

   ...

+  def handle_info({:measurement_inserted, new_measurement}, socket) do
+    # TODO: do something
+  end

Throttle incoming PubSub messages

I want to refresh the real-time dashboard based on incoming PubSub messages, but at the same time I want to control how often I refresh the dashboard no matter how fast PubSub messages are coming in. In other words, I do not want to overwhelm my LiveView rendering in case messages are coming in ridiculously fast.

So I calculate the time when I want to do the refresh next based on refresh_interval value, ignoring any messages until will_refresh_at has elapsed. I use Timex library so the time-related calculation is human-readable.

   def handle_info({:measurement_inserted, new_measurement}, socket) do
-    # TODO: do something
+    if refresh_interval_elapsed?(socket) do
+      {:noreply, assign(socket, last_measurement: new_measurement)}
+    else
+      {:noreply, socket}
+    end
   end
+
+  # Check if the refresh interval has elapsed. (next_refresh >= now)
+  defp refresh_interval_elapsed?(socket) do
+    next_refresh = DateTime.add(socket.assigns.last_measurement.measured_at, socket.assigns.refresh_interval)
+
+    case DateTime.compare(DateTime.utc_now(), next_refresh) do
+      :gt -> true
+      :eq -> true
+      _ -> false
+    end
+  end

That's it!