defmodule Phoenix.SocketClient.Channel.Helpers do @moduledoc """ Helper functions for Phoenix.SocketClient.Channel GenServer callbacks. This module contains the actual implementation logic for channel callbacks, extracted from the `__using__` macro to improve code maintainability. """ alias Phoenix.SocketClient.Channel.State alias Phoenix.SocketClient.{Message, Telemetry} @doc """ Initializes a channel process. """ def init_impl({sup_pid, socket_pid, topic, params, registry_name}) do Registry.register(registry_name, topic, self()) {:ok, %State{ sup_pid: sup_pid, socket_pid: socket_pid, topic: topic, params: params, registry_name: registry_name }} end @doc """ Handles the :join call. """ def handle_join_call( {_pid, _ref} = from, %State{sup_pid: sup_pid, topic: topic, params: params} = state ) do message = Message.join(topic, params) push = Phoenix.SocketClient.push(sup_pid, message) {:noreply, %State{ state | join_ref: push.ref, caller: elem(from, 0), pushes: [{from, push} | state.pushes], join_start_time: System.monotonic_time() }} end @doc """ Handles the :leave call. """ def handle_leave_call( _from, %{sup_pid: sup_pid, socket_pid: _socket_pid, topic: topic} = state ) do Phoenix.SocketClient.update_channel_status(sup_pid, self(), topic, :leaving) message = Message.leave(topic) _push = Phoenix.SocketClient.push(sup_pid, message) {:stop, :normal, :ok, %{state | leave_start_time: System.monotonic_time()}} end @doc """ Handles the {:push, event, payload} call. """ def handle_push_call( {event, payload}, from, %State{sup_pid: sup_pid, topic: topic} = state ) do message = %Message{ topic: topic, event: event, payload: payload, ref: Message.generate_ref(), join_ref: state.join_ref } push = Phoenix.SocketClient.push(sup_pid, message) Telemetry.message_sent(self(), topic, event, payload) {:noreply, %State{state | pushes: [{from, push} | state.pushes]}} end @doc """ Handles the :get_topic call. """ def handle_get_topic_call(_from, state) do {:reply, state.topic, state} end @doc """ Handles the {:push, event, payload} cast. """ def handle_push_cast( {event, payload}, %{sup_pid: sup_pid, topic: topic} = state ) do message = %Message{ topic: topic, event: event, payload: payload, ref: Message.generate_ref(), channel_pid: self(), join_ref: state.join_ref } Phoenix.SocketClient.push(sup_pid, message) Telemetry.message_sent(self(), topic, event, payload) {:noreply, state} end @doc """ Handles phx_reply messages. """ def handle_phx_reply_info( %Message{event: "phx_reply", ref: ref} = msg, %State{pushes: pushes, topic: topic, join_ref: join_ref, params: params} = s, _handle_message_fun, handle_join_reply_fun \\ nil ) do {pushes, state} = case Enum.split_with(pushes, &(elem(&1, 1).ref == ref)) do {[{from_ref, _push}], pushes} -> %{"status" => status, "response" => response} = msg.payload handle_reply_status(status, ref, join_ref, topic, msg, s, params) GenServer.reply(from_ref, {status_to_atom(status), response}) new_state = if ref == join_ref && handle_join_reply_fun do {:noreply, %State{} = updated} = handle_join_reply_fun.(status_to_atom(status), response, s) updated else s end {pushes, new_state} {[], pushes} -> send(s.caller, %{msg | channel_pid: s.caller, topic: s.topic}) {pushes, s} end {:noreply, %State{state | pushes: pushes}} end defp handle_reply_status("ok", ref, join_ref, topic, msg, s, params) do if ref == join_ref do maybe_emit_join_duration(s) Phoenix.SocketClient.update_channel_status( s.sup_pid, self(), s.topic, :joined, params ) Telemetry.channel_joined(s.sup_pid, s.topic, self(), msg.payload, %{}) end Telemetry.message_received(self(), topic, "phx_reply", msg.payload) end defp handle_reply_status("error", ref, join_ref, topic, msg, s, params) do if ref == join_ref do Phoenix.SocketClient.update_channel_status( s.sup_pid, self(), s.topic, :errored, params ) Telemetry.channel_join_error(s.sup_pid, s.topic, msg.payload, %{}) end Telemetry.message_received(self(), topic, "phx_reply", msg.payload) end defp handle_reply_status(_, _, _, _, _, _, _), do: :noop defp status_to_atom("ok"), do: :ok defp status_to_atom("error"), do: :error defp status_to_atom(other) do String.to_existing_atom(other) rescue ArgumentError -> other end defp maybe_emit_join_duration(%{join_start_time: nil}), do: :ok defp maybe_emit_join_duration(%{ join_start_time: start_time, socket_pid: socket_pid, topic: topic }) do duration = System.monotonic_time() - start_time Telemetry.channel_join_duration(socket_pid, topic, duration) end @doc """ Handles regular message info. """ def handle_message_info(%Message{} = message, state, handle_message_fun) do Telemetry.message_received(self(), state.topic, message.event, message.payload) handle_message_fun.(message.event, message.payload, state) end @doc """ Handles channel termination. """ def terminate_impl( reason, %{ sup_pid: sup_pid, socket_pid: socket_pid, topic: topic, params: params, registry_name: registry_name, leave_start_time: start_time } = _state ) do if start_time do duration = System.monotonic_time() - start_time Telemetry.channel_leave_duration(socket_pid, topic, duration) end Registry.unregister(registry_name, topic) if sup_pid && topic do joined_channels = Phoenix.SocketClient.get_state(sup_pid, :joined_channels) if joined_channels do channel_data = Map.get(joined_channels, topic) if reason == :normal and (channel_data && channel_data.status != :errored) do Phoenix.SocketClient.remove_channel(sup_pid, topic) else Phoenix.SocketClient.update_channel_status(sup_pid, self(), topic, :errored, params) end Telemetry.channel_left(self(), topic, reason) end end :ok end end