Phoenix.Channel поведение

Определяет канал Phoenix.

Каналы обеспечивают двунаправленную связь с клиентами и интегрируются со слоем Phoenix.PubSub для функциональности с мягким реальным временем.

Обзор концепции см. в руководстве по каналам.

Темы и колбэки

Каждый раз при подключении к каналу нужно выбрать конкретную тему, которую вы хотите прослушивать. Тема — это просто идентификатор, но по соглашению она часто состоит из двух частей: "topic:subtopic". Подход "topic:subtopic" хорошо сочетается с Phoenix.Socket.channel/3, позволяя сопоставлять все темы, начинающиеся с заданного префикса, с помощью звёздочки (символа *) в конце шаблона темы:

channel "room:*", MyAppWeb.RoomChannel

Любая тема с префиксом "room:", поступающая в маршрутизатор, будет направлена в MyAppWeb.RoomChannel в приведённом выше примере. В колбэке join/3 каналов также можно сопоставлять темы с шаблоном, чтобы извлечь шаблон с заданной областью:

# handles the special `"lobby"` subtopic
def join("room:lobby", _payload, socket) do
  {:ok, socket}
end

# handles any other subtopic as the room ID, for example `"room:12"`, `"room:34"`
def join("room:" <> room_id, _payload, socket) do
  {:ok, socket}
end

Первый аргумент — тема, второй аргумент — карта данных, переданная клиентом, а третий аргумент — экземпляр Phoenix.Socket. socket передаётся всем колбэкам канала, поэтому изучите его модуль и документацию, чтобы узнать о его полях и способах взаимодействия с ним.

Авторизация

Чтобы отправлять и получать события PubSub в канале, клиенты должны подключиться к нему. В каналах необходимо реализовать колбэк join/3, который авторизует сокет для заданной темы. Например, можно проверить, разрешено ли пользователю подключаться к этой комнате.

Чтобы авторизовать сокет в join/3, верните {:ok, socket}. Чтобы отказать в авторизации в join/3, верните {:error, reply}.

Входящие события

После успешного подключения клиента к каналу входящие от него события обрабатываются колбэками handle_in/3 канала. В этих колбэках можно выполнить любое действие. Входящие колбэки должны возвращать socket, чтобы сохранять эфемерное состояние.

Обычно вы либо пересылаете сообщение всем слушателям с помощью broadcast!/3, либо напрямую отвечаете на событие клиента в стиле «запрос/ответ».

Данные обычных сообщений могут иметь любой сериализуемый тип и часто представляют собой карты:

def handle_in("new_msg", %{"uid" => uid, "body" => body}, socket) do
  ...
  {:reply, :ok, socket}
end

Двоичные данные передаются в виде кортежа {:binary, data}:

def handle_in("file_chunk", {:binary, chunk}, socket) do
  ...
  {:reply, :ok, socket}
end

Трансляции

События можно транслировать в тему из любой части приложения с помощью функции broadcast в endpoint:

MyAppWeb.Endpoint.broadcast!("room:13", "new_message", %{content: "hello"})

Также можно выполнять трансляцию непосредственно из каналов. Ниже приведён пример получения входящего события "new_msg" от одного клиента и трансляции сообщения всем подписчикам темы для этого сокета.

def handle_in("new_msg", %{"uid" => uid, "body" => body}, socket) do
  broadcast!(socket, "new_msg", %{uid: uid, body: body})
  {:noreply, socket}
end

Ответы

Ответы полезны, чтобы подтвердить получение сообщения клиента или вернуть результаты операции. Ответ отправляется только клиенту, подключённому к текущему процессу канала. Внутри ответа содержится ref сообщения клиента, позволяющий клиенту сопоставить полученный ответ с отправленным сообщением.

Например, представьте, что вы создаёте ресурс и отвечаете созданной записью:

def handle_in("create:post", attrs, socket) do
  changeset = Post.changeset(%Post{}, attrs)

  if changeset.valid? do
    post = Repo.insert!(changeset)
    response = MyAppWeb.PostView.render("show.json", %{post: post})
    {:reply, {:ok, response}, socket}
  else
    response = MyAppWeb.ChangesetView.render("errors.json", %{changeset: changeset})
    {:reply, {:error, response}, socket}
  end
end

Или можно просто подтвердить успешное выполнение операции:

def handle_in("create:post", attrs, socket) do
  changeset = Post.changeset(%Post{}, attrs)

  if changeset.valid? do
    Repo.insert!(changeset)
    {:reply, :ok, socket}
  else
    {:reply, :error, socket}
  end
end

В ответах также поддерживаются двоичные данные в виде кортежа {:binary, data}:

{:reply, {:ok, {:binary, bin}}, socket}

Если не нужно отправлять ответ клиенту, можно вернуть:

{:noreply, socket}

Так можно поступить, если необходимо ответить позднее; см. reply/2.

Отправка сообщений

Вызов push/3 позволяет отправить клиенту сообщение, которое не является ответом на конкретное сообщение клиента. Поскольку это не ответ, отправленное сообщение не содержит ref сообщения клиента: ему не предшествовало сообщение клиента, с которым его можно было бы связать.

Возможные сценарии использования:

• Вы автоматически сохранили документ пользователя
• Игра пользователя скоро завершится
• Настройки IoT-устройства нужно обновить

Например, можно push/3 сообщение клиенту в handle_info/3 после получения относящегося к нему сообщения PubSub.

alias Phoenix.Socket.Broadcast
def handle_info(%Broadcast{topic: _, event: event, payload: payload}, socket) do
  push(socket, event, payload)
  {:noreply, socket}
end

Данные для отправки можно передать в виде карты или помеченного кортежа {:binary, data}:

# client asks for their current rank. reply contains it, and client
# is also pushed a leader board and a badge image
def handle_in("current_rank", _, socket) do
  push(socket, "leaders", %{leaders: Game.get_leaders(socket.assigns.game_id)})
  push(socket, "badge", {:binary, File.read!(socket.assigns.badge_path)})
  {:reply, %{val: Game.get_rank(socket.assigns[:user])}, socket}
end

Обратите внимание, что в этом примере push/3 вызывается из handle_in/3; таким образом, можно, по сути, ответить на одно сообщение клиента N раз. О причинах, по которым это может быть полезно, см. reply/2.

Перехват исходящих событий

Когда событие транслируется с помощью broadcast/3, каждый подписчик канала может перехватить его и вызвать свой колбэк handle_out/3. Это позволяет настраивать данные события отдельно для каждого сокета, добавляя дополнительную информацию или условно отфильтровывая сообщение. Если событие не перехвачено с помощью Phoenix.Channel.intercept/1, сообщение отправляется непосредственно клиенту:

intercept ["new_msg", "user_joined"]

# for every socket subscribing to this topic, append an `is_editable`
# value for client metadata.
def handle_out("new_msg", msg, socket) do
  push(socket, "new_msg", Map.merge(msg,
    %{is_editable: User.can_edit_message?(socket.assigns[:user], msg)}
  ))
  {:noreply, socket}
end

# do not send broadcasted `"user_joined"` events if this socket's user
# is ignoring the user who joined.
def handle_out("user_joined", msg, socket) do
  unless User.ignoring?(socket.assigns[:user], msg.user_id) do
    push(socket, "user_joined", msg)
  end
  {:noreply, socket}
end

Завершение

При завершении вызывается колбэк канала terminate/2, которому передаются причина ошибки и сокет.

Если завершение вызвано отключением клиента, причиной будет {:shutdown, :left}. Аналогично, если завершение вызвано закрытием соединения с клиентом, причиной будет {:shutdown, :closed}.

Если какой-либо из колбэков возвращает кортеж :stop, это также вызовет завершение с причиной, указанной в кортеже.

Однако terminate/2 не вызывается в случае ошибок или завершения процесса. Такое же поведение наблюдается в абстракциях Elixir, например GenServer и других. Подобно GenServer, можно также использовать :trap_exit, чтобы гарантировать вызов terminate/2. Однако такой подход не рекомендуется.

Как правило, если нужно что-либо очистить, лучше отслеживать процесс канала и выполнять очистку из другого процесса. Все колбэки канала, включая join/3, вызываются внутри процесса канала. Поэтому self() в любом из них возвращает PID для отслеживания.

Причины выхода при остановке канала

Когда колбэки канала возвращают кортеж :stop, например:

{:stop, :shutdown, socket}
{:stop, {:error, :enoent}, socket}

второй аргумент — причина выхода, которая соответствует стандартному поведению при выходе из GenServer.

При остановке канала можно выбрать один из трёх вариантов:

• :normal — в этом случае выход не будет записан в журнал, а связанные процессы не завершатся

• :shutdown или {:shutdown, term} — в этом случае выход не будет записан в журнал, а связанные процессы завершатся с той же причиной, если только они не перехватывают сигналы выхода

• любое другое значение — в этом случае выход будет записан в журнал, а связанные процессы завершатся с той же причиной, если только они не перехватывают сигналы выхода

Подписка на внешние темы

Иногда может потребоваться программно подписать сокет на внешние темы в дополнение к внутренней socket.topic. Например, представьте систему торгов, в которой удалённый клиент динамически задаёт предпочтения для товаров, о торгах по которым он хочет получать уведомления. Вместо создания отдельного процесса канала и темы для каждого предпочтения эффективнее и проще подписать один канал на соответствующие уведомления через endpoint. Например:

defmodule MyAppWeb.Endpoint.NotificationChannel do
  use Phoenix.Channel

  def join("notification:" <> user_id, %{"ids" => ids}, socket) do
    topics = for product_id <- ids, do: "product:#{product_id}"

    {:ok, socket
          |> assign(:topics, [])
          |> put_new_topics(topics)}
  end

  def handle_in("watch", %{"product_id" => id}, socket) do
    {:reply, :ok, put_new_topics(socket, ["product:#{id}"])}
  end

  def handle_in("unwatch", %{"product_id" => id}, socket) do
    {:reply, :ok, MyAppWeb.Endpoint.unsubscribe("product:#{id}")}
  end

  defp put_new_topics(socket, topics) do
    Enum.reduce(topics, socket, fn topic, acc ->
      topics = acc.assigns.topics
      if topic in topics do
        acc
      else
        :ok = MyAppWeb.Endpoint.subscribe(topic)
        assign(acc, :topics, [topic | topics])
      end
    end)
  end
end

Примечание: вызывающая сторона должна предотвращать повторные подписки. После вызова subscribe/1 из endpoint обработка обычных сообщений Elixir в канале выполняется так же, как и обычно. Чаще всего достаточно просто переслать событие %Phoenix.Socket.Broadcast{} и его данные:

alias Phoenix.Socket.Broadcast
def handle_info(%Broadcast{topic: _, event: event, payload: payload}, socket) do
  push(socket, event, payload)
  {:noreply, socket}
end

Гибернация

Начиная с Erlang/OTP 20, каналы автоматически переходят в режим гибернации после 15_000 миллисекунд бездействия, чтобы экономить память. Это можно настроить, передав параметр :hibernate_after в use Phoenix.Channel:

use Phoenix.Channel, hibernate_after: 60_000

Также можно задать значение :infinity, чтобы полностью отключить эту функцию.

Завершение работы

Можно настроить поведение при завершении работы каждого канала, используемого приложением, задав значение :shutdown при использовании:

use Phoenix.Channel, shutdown: 5_000

По умолчанию используется значение 5_000. Поддерживаемые значения описаны в документации модуля Supervisor.

Журналирование

По умолчанию события "join" и "handle_in" каналов записываются в журнал с уровнями :info и :debug соответственно. Уровень журналирования для каждого события можно изменить или отключить журналирование событий определённого типа, задав параметры :log_join и :log_handle_in при использовании Phoenix.Channel. Например, следующая конфигурация записывает события подключения с уровнем :info, но отключает журналирование входящих событий:

use Phoenix.Channel, log_join: :info, log_handle_in: false

Обратите внимание: изменение уровня для типа события не влияет на то, что записывается в журнал, если только не задать значение false; в этом случае изменится соответствующий уровень.