Каналы Phoenix - Несколько каналов на сокет

Question

Я пишу приложение с использованием каналов Elixir для обработки событий в реальном времени. Я понимаю, что на каждого клиента будет открыто 1 сокет и может мультиплексировать по нему несколько каналов. Поэтому мое приложение является чат-приложением, в котором пользователи являются частью нескольких групповых чатов. У меня есть 1 Phoenix Channel, называемый MessageChannel, где метод join будет обрабатывать темы dyanamic.

def join("groups:" <> group_id, payload, socket) do 
.... 

Скажем, Джон присоединяется к группам/Темы A и B, а Боб только присоединиться к группе/тема B. Когда Джон посылает сообщение группы/топики, радиовещательный!/3 также отправить это сообщение Бобу слишком правильный ? Поскольку handle_in не имеет контекста того, к какой теме/группе было отправлено сообщение.

Как бы я справился с этим, чтобы Боб не получил события, которые были отправлены группе А. Я правильно ли это проектирую?

Answer 1

Потому что handle_in не имеет контекста, в отношении которого тема/группа отправили сообщение.

Когда Phoenix.Channel.broadcast/3 называется, по-видимому, это делает есть тема, связанная с сообщением (который не является очевидным из подписи). Вы можете увидеть код, начинающийся on this line of channel.ex:

def broadcast(socket, event, message) do 
    %{pubsub_server: pubsub_server, topic: topic} = assert_joined!(socket) 
    Server.broadcast pubsub_server, topic, event, message 
end 

Таким образом, когда вызов broadcast/3 производится с использованием сокета, он соответствует шаблону из текущей темы, а затем делает вызов базовой Server.broadcast/4.

(Если вам интересно, как я, это, в свою очередь, вызывает базовый PubSub.broadcast/3, который выполняет некоторую магию распространения, чтобы направить вызов на ваш настроенный сервер реализации pubsub, скорее всего, используя pg2, но я отвлекся ...)

Итак, я нашел это поведение не очевидно из прочтения Phoenix.Channel docs, но они утверждают, это явно на странице phoenixframework каналов в Incoming Events:

broadcast!/3 уведомит все присоединились к клиентам на тему этого сокета и вызова их обратные вызовы handle_out/3.

Так что это только транслируется «по теме этого сокета». Они определяют тему на той же странице:

тему - строка тему или тему: подтемы пара имен, например, «Сообщения», «сообщения: 123»

Так что в вашем примере, «Темы» на самом деле являются темой: строки подпространства подтемы: "groups:A" и "groups:B". Джон должен был бы подписаться на обе эти темы отдельно на клиенте, поэтому у вас действительно есть ссылки на два разных канала, даже если они используют один и тот же сокет. Таким образом, при условии, что вы используете яваскрипт клиент, создание канала выглядит следующим образом:

let channelA = this.socket.channel("groups:A", {}); 
let channelB = this.socket.channel("groups:B", {}); 

Затем, когда вы идете, чтобы отправить сообщение на канале от клиента, вы используете только канал, который имеет тему который получает шаблон, сопоставленный на сервере, как мы видели выше.

channelA.push(msgName, msgBody); 

Answer 2

Отказ от ответственности: я не смотрел на внутреннюю работу канала, эта информация полностью связана с моим первым опытом использования каналов в приложении.

Когда кто-то присоединяется к другой группе (на основе соответствия шаблону в вашем join/3), создается соединение по отдельному каналу (сокету). Таким образом, передача в эфир A не будет отправлять сообщения членам B только A.

Мне кажется, что модуль канала похож на GenServer, и соединение похоже на start_link, где новый сервер (процесс) (однако, только если он еще не существует).

Вы действительно можете игнорировать внутреннюю работу модуля и просто понимать, что если вы присоединитесь к каналу с другим именем, чем уже существующие, вы присоединяетесь к уникальному каналу. Вы также можете просто надеяться, что если вы транслируете на канал, только участники этого канала получат сообщение.

Например, в моем приложении у меня есть канал пользователя, к которому я хочу подключиться только к одному пользователю. Соединение выглядит как def join("agent:" <> _agent, payload, socket), где агент - это просто адрес электронной почты. Когда я транслирую сообщение на этот канал, только один агент получает сообщение. У меня также есть служебный канал, к которому присоединяются все агенты, и я передаю его, когда хочу, чтобы все агенты получили сообщение.

Надеюсь, это поможет.

Answer 3

На самом деле, маршрутизация сокета выполняется на основе, как определить ваши темы в модуле проектов сокета с channel API. Для моего Slack clone я использую три канала. У меня есть канал уровня системы для обработки обновления присутствия, канала пользователя и канала комнаты.

Любой данный пользователь подписался на 0 или 1 канал. Тем не менее, пользователи могут быть подписаны на несколько каналов.

Для сообщений, выходящих в определенную комнату, я транслирую их по каналу комнаты.

Когда я обнаруживаю непрочитанные сообщения, уведомления или значки для конкретной комнаты, я использую пользовательский канал. В каждом пользовательском канале хранится список номеров, которые пользователь подписал (они указаны на боковой панели клиента).

Хитрость все это использует API, пара каналов, в основном intercept, handle_out, My.Endpoint.subscribe и handle_info(%Broadcast{},socket).

• Я использую intercept, чтобы поймать передаваемые сообщения, которые я хочу либо игнорировать, либо манипулировать перед их отправкой.
• В канале пользователя, подписаться на события транслируются из канала номера
• При подписке вы получаете handle_info вызов с %Broadcast{} структурой, которая включает в себя тему, событие, и полезную нагрузку транслируемого сообщения.

Вот пара части моего кода:

defmodule UcxChat.UserSocket do 
    use Phoenix.Socket 
    alias UcxChat.{User, Repo, MessageService, SideNavService} 
    require UcxChat.ChatConstants, as: CC 

    ## Channels 
    channel CC.chan_room <> "*", UcxChat.RoomChannel # "ucxchat:" 
    channel CC.chan_user <> "*", UcxChat.UserChannel # "user:" 
    channel CC.chan_system <> "*", UcxChat.SystemChannel # "system:" 
    # ... 
end 

# user_channel.ex 
# ... 
intercept ["room:join", "room:leave", "room:mention", "user:state", "direct:new"] 
#... 
def handle_out("room:join", msg, socket) do 
    %{room: room} = msg 
    UserSocket.push_message_box(socket, socket.assigns.channel_id, socket.assigns.user_id) 
    update_rooms_list(socket) 
    clear_unreads(room, socket) 
    {:noreply, subscribe([room], socket)} 
    end 
    def handle_out("room:leave" = ev, msg, socket) do 
    %{room: room} = msg 
    debug ev, msg, "assigns: #{inspect socket.assigns}" 
    socket.endpoint.unsubscribe(CC.chan_room <> room) 
    update_rooms_list(socket) 
    {:noreply, assign(socket, :subscribed, List.delete(socket.assigns[:subscribed], room))} 
    end 

    # ... 
    defp subscribe(channels, socket) do 
    # debug inspect(channels), "" 
    Enum.reduce channels, socket, fn channel, acc -> 
     subscribed = acc.assigns[:subscribed] 
     if channel in subscribed do 
     acc 
     else 
     socket.endpoint.subscribe(CC.chan_room <> channel) 
     assign(acc, :subscribed, [channel | subscribed]) 
     end 
    end 
    end 
    # ... 
end 

Я также использую user_channel для всех событий, связанных с конкретным пользователем, как состояние клиента, сообщения об ошибках и т.д.