Zenn
🏣

ElixirでTCP接続したクライアントからの入力を他の全クライアントにブロードキャストするサーバを作成する

2022/03/04に公開
Elixir
tech

やろうとしていることは、表題の通りです。それを実現するに際して、最近Elixirコミュニティで話題になったmtrudel/bandit: Bandit is a pure Elixir HTTP server for Plug applicationsの基底レイヤーをなすソケット通信まわりを扱うライブラリであるmtrudel/thousand_island: Thousand Island is a pure Elixir socket serverを使って実装してみた、というのがポイントです。

実現したいこと

  1. 登場人物は以下の通り
    • サーバ（ひとつ）
    • クライアント（複数）
  2. サーバは1234番ポートでTCP接続を待ち受けているので、クライアントはそれぞれnc localhost 1234として繋ぎにいく
  3. クライアントが何かメッセージを送ると、自分以外のクライアントにそのメッセージが送られる

アニメーションGifでその模様を収録しておきました。見えにくいと思うので、画像をクリックして大きくした状態で観てください。

実装

ThousandIslandプロセスの起動

まずは、プロセスを起動する設定をApplicationモジュールに施します。ここはいつも通りの感じで。

ThousandIslandだけでなく、Registryも使います。また、Registryはkeys: :duplicateオプションで起動しているので、同じキーでregisterすると、リストとして渡した値が追加されていきます（後述の実装参照のこと）。

defmodule Broadcast.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      {ThousandIsland, port: 1234, handler_module: Broadcast.Handler},
      {Registry, keys: :duplicate, name: Broadcast.Registry}
    ]

    opts = [strategy: :one_for_one, name: Broadcast.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

ハンドラーの実装

thousand_islandの特徴として、ThousandIsland.Handlerビヘイビアを実装してあげるだけで、いい感じにTCPサーバを作れるということが挙げられます。

まず、ThousandIsland.Handlerビヘイビアの要求するhandle_connection/2を実装します。クライアントからの接続があると呼ばれる関数です。この関数内で、Registryを用いて、クライアントのソケット情報を記録しておきます。

次に、ThousandIsland.Handlerビヘイビアの要求するhandle_data/3を実装します。クライアントからメッセージが送られてくると呼ばれる関数です。クライアントからのメッセージ内容が入っているので、送ってきたクライアント以外のクライアントに、そのメッセージを送信します。

送信は、いつもの通りGenServerビヘイビアのhandle_cast/2を使ってやればOKです。Registryに記録されている接続済みクライアントのソケットを取得して、送信元以外のクライアントにメッセージを送ります。

defmodule Broadcast.Handler do
  use ThousandIsland.Handler

  def send_message(pid, from, msg) do
    GenServer.cast(pid, {:send, from, msg})
  end

  @impl ThousandIsland.Handler
  def handle_connection(socket, state) do
    Registry.register(Broadcast.Registry, "clients", socket)
    {:continue, state}
  end

  @impl ThousandIsland.Handler
  def handle_data(msg, socket, state) do
    send_message(self(), socket, msg)
    {:continue, [socket | state]}
  end

  @impl GenServer
  def handle_cast({:send, from, msg}, {socket, state}) do
    Registry.lookup(Broadcast.Registry, "clients")
    |> Enum.filter(fn {_pid, socket} ->
      socket != from
    end)
    |> Enum.each(fn {_pid, socket} ->
      ThousandIsland.Socket.send(socket, msg)
    end)

    {:noreply, {socket, state}}
  end
end

ソースコード

以下のリポジトリに置いてあります。

kentaro/broadcast: An Example Broadcast Server for TCP Connection

おわりに

簡単に使えていい感じのライブラリですね。これを使ったHTTPサーバ実装のBanditの中身も、追ってみていきたいところです。

Discussion