@spec __using__(Keyword.t()) :: Macro.t()

Declares that a module is a Slipstream socket client

Slipstream provides a use Slipstream macro that behaves similar to GenServer's use GenServer. This does a few things:

• a default implementation of child_spec/1, which is used to start the module as a GenServer. This may be overridden.
• imports for all documented functions in Slipstream and Slipstream.Socket
• a GenServer.handle_info/2 function clause which matches incoming events from the connection process and dispatches them to the various Slipstream callbacks
• a GenServer.handle_info/2 function clause which matches Slipstream commands. This is used to implement back-off retry mechanisms for reconnect/1 and rejoin/3.

This provides a familiar and sleek interface for the common case of using Slipstream: an asynchronous callback-based GenServer module.

It's not required to use this macro, though. Slipstream can be used in synchronous mode (via the await_* family of functions).

options

Options

Any options passed as the opts argument to __using__/1 are passed along to Supervisor.child_spec/2, as is done in the default child_spec/1 implementation for GenServer. Most notably, you may control the restart strategy of the client with the :restart option. See the example below.

examples

Examples

defmodule MyApp.MySocketClient do
  # one crash/shutdown/exit will permanently terminate the server
  use Slipstream, restart: :temporary

  def start_link(args) do
    Slipstream.start_link(__MODULE__, args)
  end

  ..
end

@spec collect_garbage(socket :: Slipstream.Socket.t()) :: :ok

Requests that the connection process undergoe garbage collection

If you're using Slipstream to send large messages, you may wish to flush the process memory of the connection process between large messages. This can be achieved through garbage collection.

examples

Examples

iex> collect_garbage(socket)
:ok

@spec connect!(socket :: Slipstream.Socket.t(), opts :: Keyword.t()) ::
  Slipstream.Socket.t()

Same as connect/2 but raises on configuration validation error

Note that connect!/2 will not necessarily raise an error on failure to connect. The ! only pertains to the potential for raising when the configuration is invalid.

@spec connect(socket :: Slipstream.Socket.t(), opts :: Keyword.t()) ::
  {:ok, Slipstream.Socket.t()} | {:error, NimbleOptions.ValidationError.t()}

Requests connection to the remote endpoint

opts are passed to Slipstream.Configuration.validate/1 before sending.

Note that this request for connection is asynchronous. A return value of {:ok, socket} does not mean that a connection has successfully been established.

examples

Examples

{:ok, socket} = connect(uri: "ws://localhost:4000/socket/websocket")

@spec disconnect(socket :: Slipstream.Socket.t()) :: Slipstream.Socket.t()

Requests that an open connection be closed

This function will no-op when the socket is not currently connected to any remote websocket server.

Note that you do not need to use disconnect/1 to clean up a connection. The connection process monitors the slipstream client process and will shut down when it detects that the process has terminated.

Disconnection may be awaited synchronously with await_disconnect/2

examples

Examples

@impl Slipstream
def handle_info(:chaos_monkey, socket) do
  {:ok, socket} =
    socket
    |> disconnect()
    |> await_disconnect()

  {:noreply, reconnect(socket)}
end

@spec join(
  socket :: Slipstream.Socket.t(),
  topic :: String.t(),
  params :: json_serializable()
) :: Slipstream.Socket.t()

Requests that a topic be joined in the current connection

Multiple topics may be joined by one Slipstream client, but each topic may only be joined once. Despite this, join/3 may be called on the same topic multiple times, but the result will be idempotent. The client will not request to join unless it has not yet joined that topic. In cases where you wish to begin a new session with a topic, you must first leave/2 and then join/3 again.

The request to join will not error-out if the client is not connected to a remote server. In that case, the join/3 function will act as a no-op.

A join can be awaited in a blocking fashion with await_join/3.

examples

Examples

@impl Slipstream
def handle_connect(socket) do
  {:ok, join(socket, "rooms:lobby", %{user: 1})}
end

@spec leave(socket :: Slipstream.Socket.t(), topic :: String.t()) ::
  Slipstream.Socket.t()

Requests that the given topic be left

Note that like joining, leaving is an asynchronous request and can be awaited with await_leave/3.

Also similar to join/3, leave/2 is idempotent and will not raise an error if the provided topic is not currently joined.

examples

Examples

iex> {:ok, socket} = leave(socket, "room:lobby") |> await_leave("rooms:lobby")
iex> join(socket, "rooms:specific")

@spec new_socket() :: Slipstream.Socket.t()

Creates a new socket without immediately connecting to a remote websocket

This can be useful if you do not wish to request connection with connect/2 during the init/1 callback (because the init/1 callback requires that you return a Slipstream.Socket.t/0).

examples

Examples

defmodule MySocketClient do
  use Slipstream

  ..

  @impl Slipstream
  def init(args) do
    {:ok, new_socket() |> assign(:init_args, args)}
  end

  ..
end

iex> new_socket()
#Slipstream.Socket<assigns: %{}, ...>

@spec push!(
  socket :: Slipstream.Socket.t(),
  topic :: String.t(),
  event :: String.t(),
  params :: json_serializable()
) :: push_reference()

Pushes, raising if the topic is not joined or if the channel is dead

Same as push/4, but raises in cases of failure.

This can be useful for pipeing into await_reply/2

examples

Examples

iex> {:ok, result} = push!(socket, "rooms:lobby", "msg:new", params) |> await_reply()
{:ok, %{"created?" => true}}

@spec push(
  socket :: Slipstream.Socket.t(),
  topic :: String.t(),
  event :: String.t(),
  params :: json_serializable() | {:binary, binary()},
  timeout :: timeout()
) :: {:ok, push_reference()} | {:error, reason :: term()}

Requests that a message be pushed on the websocket connection

A channel has the ability to reply directly to a message, but this reply is asynchronous. Handle replies using the handle_reply/3 callback or by awaiting them synchronously with await_reply/2.

Although this request to the remote server is asynchronous, the call to the transport process to transmit the push is synchronous and will exert back-pressure on calls to push/4, as push/4 blocks until the message has been sent by the transport.

If you are pushing especially large messages, you may need to adjust the timeout argument so that the GenServer call does not exit with :timeout. The default value is 5_000 msec (5 seconds).

A phoenix channel may decide to reply to a message sent with push/2. In order to link push requests to their replies, store the ref string returned from the call to push/4 and match on it in handle_reply/3.

examples

Examples

@impl Slipstream
def handle_join(:success, _response, state) do
  {:ok, hello_request} = push(socket, "rooms:lobby", "new:msg", %{user: 1, body: "hello"})

  {:ok, Map.put(state, :hello_request, hello_request)}
end

@impl Slipstream
def handle_reply(ref, reply, %{hello_request: ref} = state) do
  IO.inspect(reply, label: "nice, a response.")

  {:ok, state}
end

@spec reconnect(socket :: Slipstream.Socket.t()) ::
  {:ok, Slipstream.Socket.t()} | {:error, :no_config | :not_connected}

Request reconnection given the last-used connection configuration

Note that when reconnect/1 is used to re-connect instead of connect/2 (or connect!/2), the slipstream process will attempt to reconnect with a retry mechanism with backoff. The process will wait an interval between reconnection attempts following the list of milliseconds provided in the :reconnect_after_msec key of configuration passed to connect/2 (or connect!/2).

The handle_disconnect/2 callback will be invoked for each failure to re-connect, however, so an implementation of that callback which will simply retry with backoff can be achieved like so:

@impl Slipstream
def handle_disconnect(_reason, socket) do
  case reconnect(socket) do
    {:ok, socket} -> {:ok, socket}
    {:error, reason} -> {:stop, reason, socket}
  end
end

reconnect/1 may return an :error tuple in the case that the socket passed does not contain any connection information (which is added to the socket with connect/2 or connect!/2), or if the socket is currently connected. For a reconnect/1 call without configuration, the return pattern is {:error, :no_config}, and for a socket that is already connected, the pattern is {:error, :connected}.

A reconnect may be awaited with await_connect/2.

@spec rejoin(
  socket :: Slipstream.Socket.t(),
  topic :: String.t(),
  params :: json_serializable()
) :: {:ok, Slipstream.Socket.t()} | {:error, :never_joined}

Requests that the specified topic be joined again

If params is not provided, the previously used value will be sent.

In the case that the specified topic has not been joined before, rejoin/3 will return {:error, :never_joined}.

Note that a rejoin may be awaited with await_join/3.

dealing-with-crashes

Dealing with crashes

When attempting to re-join a disconnected topic with rejoin/3, the Slipstream process will attempt to use backoff governed by the :rejoin_after_msec list in configuration passed to connect/2 (or connect!/2).

The handle_topic_close/3 callback will be invoked with the for each crash, however, so a minimal implementation of that callback which achieves the backoff retry is like so:

examples

Examples

@impl Slipstream
def handle_topic_close(topic, _reason, socket) do
  {:ok, _socket} = rejoin(socket, topic)
end

@spec start_link(module(), any(), GenServer.options()) :: GenServer.on_start()

Starts a slipstream client process

This function delegates to GenServer.start_link/3, so all options passable to that function are passable here as well. Most notably, you may name your slipstream clients using the same naming rules as GenServers, as in the example below.

examples

Examples

defmodule MySlipstreamClient do
  use Slipstream

  def start_link(args) do
    Slipstream.start_link(__MODULE__, args, name: __MODULE__)
  end

  ..
end