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

Реализация сокета, которая мультиплексирует сообщения через каналы.

Phoenix.Socket используется как модуль для установки соединения между клиентом и сервером. После установки соединения начальное состояние сохраняется в структуре Phoenix.Socket.

Один и тот же сокет можно использовать для получения событий от разных транспортов. Phoenix поддерживает параметры websocket и longpoll при вызове Phoenix.Endpoint.socket/3 в эндпоинте. По умолчанию задано websocket, а longpoll также можно настроить явно.

socket "/socket", MyAppWeb.Socket, websocket: true, longpoll: false

Приведённая выше команда позволяет устанавливать входящие соединения с сокетом через WebSocket-соединение. Входящие и исходящие события направляются в каналы по топику:

channel "room:lobby", MyAppWeb.LobbyChannel

Дополнительную информацию о каналах см. в разделе Phoenix.Channel.

Поведение сокета

Обработчики сокетов подключаются в эндпоинтах и должны определять два обратных вызова:

• connect/3 — получает параметры сокета и, если есть, сведения о соединении, а также аутентифицирует соединение. Должен возвращать структуру Phoenix.Socket, часто с пользовательскими assigns

• id/1 — получает сокет, возвращённый connect/3, и возвращает идентификатор этого соединения в виде строки. id используется для идентификации соединений сокетов, часто принадлежащих определённому пользователю, что позволяет принудительно разрывать соединения. Для сокетов, не требующих аутентификации, можно вернуть nil

Примеры

defmodule MyAppWeb.UserSocket do
  use Phoenix.Socket

  channel "room:*", MyAppWeb.RoomChannel

  def connect(params, socket, _connect_info) do
    {:ok, assign(socket, :user_id, params["user_id"])}
  end

  def id(socket), do: "users_socket:#{socket.assigns.user_id}"
end

# Disconnect all user's socket connections and their multiplexed channels
MyAppWeb.Endpoint.broadcast("users_socket:" <> user.id, "disconnect", %{})

Поля сокета

• :id — строковый идентификатор сокета
• :assigns — map с assigns сокета, значение по умолчанию: %{}
• :channel — текущий модуль канала
• :channel_pid — pid канала
• :endpoint — модуль эндпоинта, из которого был создан этот сокет, например: MyAppWeb.Endpoint
• :handler — модуль сокета, из которого был создан этот сокет, например: MyAppWeb.UserSocket
• :joined — присоединился ли сокет к каналу
• :join_ref — ref, отправленный клиентом при присоединении
• :ref — последний ref, отправленный клиентом
• :pubsub_server — зарегистрированное имя pubsub-сервера сокета
• :topic — строковый топик, например "room:123"
• :transport — идентификатор транспорта, используемый для ведения журнала
• :transport_pid — pid процесса транспорта сокета
• :serializer — сериализатор сообщений сокета

Использование параметров

Для use Phoenix.Socket принимаются следующие параметры:

• :log — уровень ведения журнала по умолчанию для действий сокета. По умолчанию — :info. Можно задать false, чтобы отключить ведение журнала

• :partitions — каждый канал запускается под супервизором. Этот параметр задаёт количество супервизоров, запускаемых для обработки каналов. По умолчанию используется количество ядер.

Сборка мусора

Можно принудительно запустить сборку мусора в процессе транспорта после обработки больших сообщений. Например, чтобы запустить её из каналов, выполните:

send(socket.transport_pid, :garbage_collect)

Также можно настроить сокет эндпоинта на более частый запуск полной сборки мусора, задав параметр :fullsweep_after для WebSocket. Дополнительную информацию см. в разделе Phoenix.Endpoint.socket/3.

Обмен данными между клиентом и сервером

Кодирование данных сервера и декодирование данных клиента выполняются согласно сериализатору, определённому в Phoenix.Socket.Serializer. По умолчанию для передачи сообщений клиентам и от них используется кодирование JSON.

Функция сериализатора decode! должна возвращать Phoenix.Socket.Message, который передаётся каналам, за исключением следующих случаев:

• "heartbeat" — события в топике "phoenix": следует просто отправить ответ OK
• "phx_join" для любого топика — следует присоединиться к топику
• "phx_leave" для любого топика — следует покинуть топик

У каждого сообщения также есть поле ref, используемое для отслеживания ответов.

Сервер может отправлять сообщения или ответы. Для сообщений ref уникально идентифицирует сообщение. Для ответов ref совпадает с ref исходного сообщения. Оба типа данных также содержат join_ref, который уникально идентифицирует текущий канал, к которому выполнено присоединение.

Реализация Phoenix.Socket также может отправлять специальные сообщения и ответы:

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

• "phx_close" — канал был корректно закрыт

В Phoenix входят реализации WebSocket и long polling на JavaScript, взаимодействующие с Phoenix.Socket. Их можно использовать как образец при создании собственных клиентов.

Пользовательские сокеты и транспорты

Дополнительную информацию о создании собственного сокета, не использующего каналы, или собственных транспортов для взаимодействия с другими сокетами см. в документации Phoenix.Socket.Transport.

Пользовательские каналы

Любой модуль можно указать как канал, если он реализует функцию child_spec/1. Функция child_spec/1 получает вызывающий процесс в качестве аргумента и должна возвращать спецификацию дочернего процесса, которая запускает процесс.

После инициализации процесс получит следующее сообщение:

{Phoenix.Channel, auth_payload, from, socket}

Пользовательская реализация канала ОБЯЗАНА вызвать GenServer.reply(from, {:ok | :error, reply_payload}) во время инициализации и передать пользовательский reply_payload, который будет отправлен клиенту в качестве ответа. Если этого не сделать, сокет будет заблокирован навсегда.

Пользовательский канал получает структуры Phoenix.Socket.Message в виде обычных сообщений от транспорта. Ответы на эти сообщения и пользовательские сообщения можно в любой момент отправить сокету, создав соответствующие структуры Phoenix.Socket.Reply и Phoenix.Socket.Message, закодировав их с помощью сериализатора и передав сериализованный результат транспорту.

Например, для обработки сообщений "phx_leave", которые рекомендуется обрабатывать во всех реализациях каналов, можно использовать следующий код:

def handle_info(
      %Message{topic: topic, event: "phx_leave"} = message,
      %{topic: topic, serializer: serializer, transport_pid: transport_pid} = socket
    ) do
  send transport_pid, serializer.encode!(build_leave_reply(message))
  {:stop, {:shutdown, :left}, socket}
end

Особое сообщение, доставляемое всем каналам, — это Broadcast с событием "phx_drain", отправляемое при остановке сокета во время завершения работы приложения. Обычно его обрабатывают, отправляя транспорту сообщение drain, которое приводит к его остановке:

def handle_info(
      %Broadcast{event: "phx_drain"},
      %{transport_pid: transport_pid} = socket
    ) do
  send(transport_pid, :socket_drain)
  {:stop, {:shutdown, :draining}, socket}
end

Кроме того, мы рекомендуем всем каналам отслеживать transport_pid на init и завершаться, если завершается транспорт. Мы также рекомендуем преобразовывать причины завершения :normal (обычно возникающие из-за закрытия сокета) в {:shutdown, :closed}, чтобы гарантировать разрыв связей при завершении канала (завершение :normal не разрывает связи):

def handle_info({:DOWN, _, _, transport_pid, reason}, %{transport_pid: transport_pid} = socket) do
  reason = if reason == :normal, do: {:shutdown, :closed}, else: reason
  {:stop, reason, socket}
end

Уровень сокета рассматривает завершение любого процесса как ошибку, если перед остановкой сокету не было отправлено сообщение {:socket_close, pid, reason}.

Пользовательские реализации каналов нельзя тестировать с помощью Phoenix.ChannelTest.