phoenix

1.2.0

1.8.9 1.8.8 1.8.7 1.8.6 1.8.5 1.8.4 1.8.3 1.8.2 1.8.1 1.8.0 1.8.0-rc.4 1.8.0-rc.3 1.8.0-rc.2 1.8.0-rc.1 1.8.0-rc.0 1.7.24 1.7.23 1.7.22 1.7.21 1.7.20 1.7.19 1.7.18 1.7.17 1.7.16 1.7.15 1.7.14 1.7.13 1.7.12 1.7.11 1.7.10 1.7.9 1.7.8 1.7.7 1.7.6 1.7.5 1.7.4 1.7.3 1.7.2 1.7.1 1.7.0 1.7.0-rc.3 1.7.0-rc.2 1.7.0-rc.1 1.7.0-rc.0 1.6.17 1.6.16 1.6.15 1.6.14 1.6.13 1.6.12 1.6.11 1.6.10 1.6.9 1.6.8 1.6.7 1.6.6 1.6.5 1.6.4 1.6.3 1.6.2 1.6.1 1.6.0 1.6.0-rc.1 1.6.0-rc.0 1.5.15 1.5.14 1.5.13 1.5.12 1.5.11 1.5.10 1.5.9 1.5.8 1.5.7 1.5.6 1.5.5 1.5.4 1.5.3 1.5.2 1.5.1 1.5.0 1.5.0-rc.0 1.4.18 1.4.17 1.4.16 1.4.15 1.4.14 1.4.13 1.4.12 1.4.11 1.4.10 1.4.9 1.4.8 1.4.7 1.4.6 1.4.5 1.4.4 1.4.3 1.4.2 1.4.1 1.4.0 1.4.0-rc.3 1.4.0-rc.2 1.4.0-rc.1 1.4.0-rc.0 1.3.5 1.3.4 1.3.3 1.3.2 1.3.1 1.3.0 1.3.0-rc.3 1.3.0-rc.2 1.3.0-rc.1 1.3.0-rc.0 1.2.5 1.2.4 1.2.3 1.2.2 1.2.1 1.2.0 1.2.0-rc.1 1.2.0-rc.0 1.1.9 1.1.8 1.1.7 1.1.6 1.1.5 1.1.4 1.1.3 1.1.2 1.1.1 1.1.0 1.0.6 1.0.5 1.0.4 1.0.3 1.0.2 1.0.1 1.0.0 0.17.1 0.17.0 0.16.1 0.16.0 0.15.0 0.14.0 0.13.1 0.13.0 0.12.0 0.11.0 0.10.0 0.9.0 0.8.0 0.7.2 0.7.1 0.7.0 0.6.2 0.6.1 0.6.0 0.5.0 0.4.1 0.4.0 0.3.1 0.3.0 0.2.11 0.2.10 0.2.9 0.2.8 0.2.7 0.2.6 0.2.5 0.2.4 0.2.3 0.2.2 0.2.1 0.2.0 0.1.0

Productive. Reliable. Fast. A productive web framework that does not compromise speed or maintainability.

Security advisory: This version has known vulnerabilities. View advisories

HexDocs

Current section

Files

Jump to

Readme 176 Versions 4 Dependencies Dependants 5 Advisories Files Activity

phoenix lib phoenix transports long_poll_server.ex

Raw

lib/phoenix/transports/long_poll_server.ex

defmodule Phoenix.Transports.LongPoll.Supervisor do
  @moduledoc false

  use Supervisor

  def start_link do
    Supervisor.start_link(__MODULE__, [], name: __MODULE__)
  end

  def init([]) do
    children = [
      worker(Phoenix.Transports.LongPoll.Server, [], restart: :temporary)
    ]
    supervise(children, strategy: :simple_one_for_one)
  end
end

defmodule Phoenix.Transports.LongPoll.Server do
  @moduledoc false

  use GenServer

  alias Phoenix.PubSub
  alias Phoenix.Socket.Transport
  alias Phoenix.Socket.Broadcast
  alias Phoenix.Socket.Message

  @doc """
  Starts the Server.

    * `socket` - The `Phoenix.Socket` struct returend from `connect/2`
      of the socket handler.
    * `window_ms` - The longpoll session timeout, in milliseconds

  If the server receives no message within `window_ms`, it terminates
  and clients are responsible for opening a new session.
  """
  def start_link(endpoint, handler, transport_name, transport,
                 serializer, params, window_ms, priv_topic) do
    GenServer.start_link(__MODULE__, [endpoint, handler, transport_name, transport,
                                      serializer, params, window_ms, priv_topic])
  end

  ## Callbacks

  def init([endpoint, handler, transport_name, transport,
            serializer, params, window_ms, priv_topic]) do
    Process.flag(:trap_exit, true)

    case Transport.connect(endpoint, handler, transport_name, transport, serializer, params) do
      {:ok, socket} ->
        state = %{buffer: [],
                  socket: socket,
                  channels: %{},
                  channels_inverse: %{},
                  window_ms: trunc(window_ms * 1.5),
                  pubsub_server: socket.endpoint.__pubsub_server__(),
                  priv_topic: priv_topic,
                  last_client_poll: now_ms(),
                  client_ref: nil}

        if socket.id, do: PubSub.subscribe(state.pubsub_server, socket.id, link: true)
        :ok = PubSub.subscribe(state.pubsub_server, priv_topic, link: true)

        schedule_inactive_shutdown(state.window_ms)

        {:ok, state}
      :error ->
        :ignore
    end
  end

  def handle_call(:stop, _from, state), do: {:stop, :shutdown, :ok, state}

  # Handle client dispatches
  def handle_info({:dispatch, client_ref, msg, ref}, state) do
    msg
    |> Transport.dispatch(state.channels, state.socket)
    |> case do
      {:joined, channel_pid, reply_msg} ->
        broadcast_from!(state, client_ref, {:dispatch, ref})
        new_state = %{state | channels: Map.put(state.channels, msg.topic, channel_pid),
                              channels_inverse: Map.put(state.channels_inverse, channel_pid, {msg.topic, msg.ref})}
        publish_reply(reply_msg, new_state)

      {:reply, reply_msg} ->
        broadcast_from!(state, client_ref, {:dispatch, ref})
        publish_reply(reply_msg, state)

      :noreply ->
        broadcast_from!(state, client_ref, {:dispatch, ref})
        {:noreply, state}

      {:error, reason, error_reply_msg} ->
        broadcast_from!(state, client_ref, {:error, reason, ref})
        publish_reply(error_reply_msg, state)
    end
  end

  # Detects disconnect broadcasts and shuts down
  def handle_info(%Broadcast{event: "disconnect"}, state) do
    {:stop, {:shutdown, :disconnected}, state}
  end

  def handle_info({:EXIT, channel_pid, reason}, state) do
    case Map.get(state.channels_inverse, channel_pid) do
      nil ->
        {:stop, {:shutdown, :pubsub_server_terminated}, state}
      {topic, join_ref} ->
        new_state = delete(state, topic, channel_pid)
        publish_reply(Transport.on_exit_message(topic, join_ref, reason), new_state)
    end
  end

  def handle_info({:subscribe, client_ref, ref}, state) do
    broadcast_from!(state, client_ref, {:subscribe, ref})
    {:noreply, state}
  end

  def handle_info({:flush, client_ref, ref}, state) do
    case state.buffer do
      [] ->
        {:noreply, %{state | client_ref: {client_ref, ref}, last_client_poll: now_ms()}}
      buffer ->
        broadcast_from!(state, client_ref, {:messages, Enum.reverse(buffer), ref})
        {:noreply, %{state | client_ref: nil, last_client_poll: now_ms(), buffer: []}}
    end
  end

  def handle_info(:shutdown_if_inactive, state) do
    if now_ms() - state.last_client_poll > state.window_ms do
      {:stop, {:shutdown, :inactive}, state}
    else
      schedule_inactive_shutdown(state.window_ms)
      {:noreply, state}
    end
  end

  def handle_info(%Message{} = msg, state) do
    publish_reply(msg, state)
  end

  def terminate(_reason, _state) do
    :ok
  end

  defp broadcast_from!(state, client_ref, msg) when is_binary(client_ref),
    do: PubSub.broadcast_from!(state.pubsub_server, self, client_ref, msg)
  defp broadcast_from!(_state, client_ref, msg) when is_pid(client_ref),
    do: send(client_ref, msg)

  defp publish_reply(msg, state) do
    msg = state.socket.serializer.encode!(msg)

    case state.client_ref do
      {client_ref, ref} ->
        broadcast_from!(state, client_ref, {:now_available, ref})
      nil ->
        :ok
    end

    {:noreply, %{state | buffer: [msg | state.buffer]}}
  end

  defp now_ms, do: System.system_time(:milli_seconds)

  defp schedule_inactive_shutdown(window_ms) do
    Process.send_after(self, :shutdown_if_inactive, window_ms)
  end

  defp delete(state, topic, channel_pid) do
    case Map.fetch(state.channels, topic) do
      {:ok, ^channel_pid} ->
        %{state | channels: Map.delete(state.channels, topic),
                  channels_inverse: Map.delete(state.channels_inverse, channel_pid)}
      {:ok, _newer_pid} ->
        %{state | channels_inverse: Map.delete(state.channels_inverse, channel_pid)}
    end
  end
end

Operator	Description	Example
name:	Match package name. Supports * wildcards and repo/package form	name:phx* or name:hexpm/phoenix
description:	Full-text search of package descriptions	description:auth
depends:	Packages depending on a given package. Supports repo:package form	depends:ecto or depends:hexpm:ecto
build_tool:	Filter by build tool	build_tool:mix
updated_after:	Packages updated after an ISO8601 datetime	updated_after:2025-01-01T00:00:00Z
extra:	Match custom metadata (key,value). Nested keys are separated by commas	extra:license,MIT