Elixir, Phoenix, LiveView & PubSubでリアルタイムにページ更新

最近、趣味で[プログラミング言語Elixir](https://elixir-lang.org/)、[IoTプラットフォームNerves](https://www.nerves-project.org/)、[WebフレームワークPhoenix](https://phoenixframework.org/)を使用して、モダンで快適なIoT開発を楽しんでいます。

[autoracex](https://autoracex.connpass.com/)主催者@torifukukaiouさんの[記事](https://qiita.com/torifukukaiou/items/5876bc4576e7b7991347)を参考に自宅の温度と湿度をリアルタイムで監視できるシステムを作りました。LiveViewについては[The Pragmatic StudioのLiveViewオンラインコース](https://online.pragmaticstudio.com/courses/liveview-pro/modules/16)で勉強しています。

https://qiita.com/torifukukaiou/items/5876bc4576e7b7991347

APIサーバーがセンサー測定値を受け入れ、リアルタイムダッシュボードを見ているすべてのユーザーに対して、更新情報をプッシュします。

今日は、[Phoenix LiveView](https://hexdocs.pm/phoenix_live_view/Phoenix.LiveView.html)と[Phoenix PubSub](https://hexdocs.pm/phoenix_pubsub/Phoenix.PubSub.html)使用してのリアルタイムにページ更新することについてまとめます。

[4/3(土) 00:00〜 4/5(月) 23:59開催のautoracex #21](https://autoracex.connpass.com/event/209286/)での成果です。

[English edition](https://dev.to/mnishiguchi/real-time-monitoring-using-phoenix-liveview-and-pubsub-37g8)

[![hello-nerves-2](https://user-images.githubusercontent.com/7563926/113411295-6508b380-9383-11eb-81ef-942e0999d0cd.gif)](https://dev.to/Example/iot-development-using-rapberry-pi-and-elixir-iij)

## メッセージ購読および配信するため関数を実装

まず、特定のコンテキストモジュールでメッセージを購読および配信するための関数を準備します。トピックとして`inspect(__MODULE__)`を使用しています。そうすることで、トピック名を一意にできますし、トピック名の決定やトピック名の発見にかかる時間を節約できます。

```diff_elixir
 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
+
```

## 必要に応じてメッセージ配信

必要に応じてメッセージを配信します。次の例では、新しく挿入されたレコードをすべての購読者に通知します。

```diff_elixir
 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
```

## `LiveView`接続完了後にメッセージ購読

```diff_elixir
 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_info/2`を実装する必要があります。パターンマッチしてイベントごとに実装します。

```diff
 defmodule ExampleWeb.EnvironmentLive do
   use ExampleWeb, :live_view

   ...

+  def handle_info({:measurement_inserted, new_measurement}, socket) do
+    {:noreply, assign(socket, last_measurement: new_measurement)}
+  end
```

## 着信メッセージの[スロットリング](https://docs.developer.amazonservices.com/ja_JP/dev_guide/DG_Throttling.html)

着信メッセージによりダッシュボードをリアルタイムに更新するのですが、（すべてのメッセージに対して処理するのではなく）予め`LiveView`側で指定した周期で更新したいです。そうすることでリクエストが殺到したときに`LiveView`のレンダリングが追いつかなくなることを回避できます。

```diff_elixir
   def handle_info({:measurement_inserted, new_measurement}, socket) do
-    {:noreply, assign(socket, last_measurement: new_measurement)}
+    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
```

（IPごとに）APIエンドポイントに対して[grempe/ex_rated](https://github.com/grempe/ex_rated)を用いてレート制限もかけようと考えています。

```elixir
# enforce a rate limit of no more than 5 calls in 10 seconds
iex> ExRated.check_rate("my-rate-limited-api", 10_000, 5)
{:ok, 1}
iex> ExRated.check_rate("my-rate-limited-api", 10_000, 5)
{:ok, 2}
iex> ExRated.check_rate("my-rate-limited-api", 10_000, 5)
{:ok, 3}
iex> ExRated.check_rate("my-rate-limited-api", 10_000, 5)
{:ok, 4}
iex> ExRated.check_rate("my-rate-limited-api", 10_000, 5)
{:ok, 5}
iex> ExRated.check_rate("my-rate-limited-api", 10_000, 5)
{:error, 5}
```

おしまい。