defmodule Absinthe.GraphqlWS.Socket do @moduledoc """ This module is used by a custom websocket, which can then handle connections from a client implementing the [GraphQL over WebSocket protocol](https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md) ## Options * `schema` - required - The Absinthe schema for the current application (example: `MyAppWeb.Schema`) * `keepalive` - optional - Interval in milliseconds to send `:ping` control frames over the websocket. Defaults to `30_000` (30 seconds). * `pipeline` - optional - A `{module, function}` tuple defining how to generate an Absinthe pipeline for each incoming message. Defaults to `{Absinthe.GraphqlWS.Socket, :absinthe_pipeline}`. ## Example defmodule MyAppWeb.GraphqlSocket do use Absinthe.GraphqlWS.Socket, schema: MyAppWeb.Schema def handle_message(_msg, socket) do {:ok, socket} end end """ alias Absinthe.GraphqlWS.Socket @default_keepalive 30_000 @enforce_keys ~w[absinthe connect_info endpoint handler keepalive pubsub]a defstruct [ :absinthe, :connect_info, :endpoint, :handler, :keepalive, :pubsub, assigns: %{}, subscriptions: %{} ] @typedoc """ A socket that holds information necessary for parsing incoming messages as well as outgoing subscription data. """ @type t() :: %Socket{ absinthe: map(), assigns: map(), connect_info: map(), endpoint: module(), keepalive: integer(), subscriptions: map() } @typedoc """ Opcode atoms for messages handled by `handle_control/2`. Used by server-side keepalive messages. """ @type control() :: :ping | :pong @typedoc """ Opcode atoms for messages returned by `handle_in/2`. """ @type opcode() :: :text | :binary | control() @typedoc """ Valid replies from `Absinthe.GraphqlWS.Transport.handle_in/2` """ @type reply() :: {:ok, t()} | {:reply, :ok, {opcode(), term()}, t()} | {:reply, :error, {opcode(), term()}, t()} | {:stop, term(), t()} @typedoc """ Valid replies from `c:handle_message/2` """ @type response() :: {:ok, t()} | {:push, {opcode(), term()}, t()} | {:stop, term(), t()} @doc """ Handles messages that are sent to this process through `send/2`, which have not been caught by the default implementation. ## Example def handle_message({:thing, thing}, socket) do {:ok, assign(socket, :thing, thing)} end def handle_message(_msg, socket) do {:ok, socket} end """ @callback handle_message(params :: term(), t()) :: Socket.response() @optional_callbacks handle_message: 2 @spec __after_compile__(any(), any()) :: :ok def __after_compile__(env, _bytecode) do opts = Module.get_attribute(env.module, :graphql_ws_socket_opts) unless Keyword.has_key?(opts, :schema) do :elixir_errors.erl_warn(env.line, env.file, "#{env.module} must specify `:schema` when using Absinthe.GraphqlWS.Socket") end :ok end defmacro __using__(opts) do quote do @graphql_ws_socket_opts unquote(opts) @after_compile Absinthe.GraphqlWS.Socket import Absinthe.GraphqlWS.Util alias Absinthe.GraphqlWS.Socket @behaviour Phoenix.Socket.Transport @behaviour Absinthe.GraphqlWS.Socket @doc false @impl Phoenix.Socket.Transport def child_spec(opts) do Socket.__child_spec__(__MODULE__, opts, @graphql_ws_socket_opts) end @doc false @impl Phoenix.Socket.Transport def connect(transport) do Socket.__connect__(__MODULE__, transport, @graphql_ws_socket_opts) end @doc false @impl Phoenix.Socket.Transport def init(socket) do if socket.keepalive > 0, do: Process.send_after(self(), :keepalive, socket.keepalive) {:ok, socket} end @doc false @impl Phoenix.Socket.Transport def handle_control(message, socket), do: Absinthe.GraphqlWS.Transport.handle_control(message, socket) @doc false @impl Phoenix.Socket.Transport def handle_in(message, socket), do: Absinthe.GraphqlWS.Transport.handle_in(message, socket) @doc false @impl Phoenix.Socket.Transport def handle_info(message, socket), do: Absinthe.GraphqlWS.Transport.handle_info(message, socket) @doc false @impl Phoenix.Socket.Transport def terminate(message, socket), do: Absinthe.GraphqlWS.Transport.terminate(message, socket) defoverridable terminate: 2 end end @doc false def new(attrs \\ []), do: __struct__(attrs) @doc """ Provides a stub implementation that allows the socket to start. Phoenix.Socket.Transport expects a child spec that starts a process, so we do so with a noop Task. """ def __child_spec__(_module, _opts, _socket_opts) do %{id: Task, start: {Task, :start_link, [fn -> :ok end]}, restart: :transient} end @doc """ When a client connects to this websocket, this function is called to initialize the socket. """ def __connect__(module, socket, options) do absinthe_pipeline = Keyword.get(options, :pipeline, {__MODULE__, :absinthe_pipeline}) pubsub = socket.endpoint.config(:pubsub_server) schema = Keyword.fetch!(options, :schema) keepalive = Keyword.get(options, :keepalive, @default_keepalive) absinthe_config = %{ opts: [ context: %{ pubsub: socket.endpoint } ], pipeline: absinthe_pipeline, schema: schema } {:ok, Socket.new( absinthe: absinthe_config, connect_info: socket.connect_info, endpoint: socket.endpoint, handler: module, keepalive: keepalive, pubsub: pubsub )} end @doc """ Provides the default absinthe pipeline """ def absinthe_pipeline(schema, options) do schema |> Absinthe.Pipeline.for_document(options) end end