RTMP.Handler behaviour (rtmp v0.1.0)

The behaviour you implement to receive RTMP publishes.

One handler state exists per connection. Callbacks run inside the connection's own process, in order, and the next chunk of network data is not read until a callback returns — so a slow callback slows the publisher down instead of making memory grow (see "Backpressure" in the README).

Lifecycle

init/2              TCP connection accepted (before the RTMP handshake)
handle_connect/2    client sent `connect`
handle_publish/3    client sent `publish` — accept or reject the stream key
handle_metadata/2   client sent `onMetaData` (zero or more times)
handle_frame/2      every audio/video frame
handle_disconnect/2 the connection is gone, for whatever reason

handle_publish/3 and handle_frame/2 are required. The others are optional; use RTMP.Handler defines defaults (accept everything, ignore metadata) that you can override.

There is deliberately no default for handle_publish/3: deciding which stream keys are valid is the one thing every server must do on purpose.

Disconnect reasons

handle_disconnect/2 receives one of:

:closed — the client closed the TCP connection.
:unpublished — the client sent deleteStream for the published stream.
:idle_timeout — no data within idle_timeout after the handshake.
:handshake_timeout — a handshake step took longer than handshake_timeout.
{:rejected, reason} — your handle_connect/2 or handle_publish/3 rejected the client, the client asked to play (not supported), or it published a codec this library rejects ({:unsupported_codec, ...}).
{:disconnected, reason} — your handle_metadata/2 or handle_frame/2 returned {:disconnect, reason, state}, or RTMP.disconnect/2 was called.
{:protocol_error, reason} — the client broke the protocol or a limit in RTMP.Limits.
{:handler_error, exception} — one of your callbacks raised. The connection is closed cleanly and the error is logged.
:shutdown — the listener is shutting down.
{:socket_error, reason} — the socket failed.

handle_disconnect/2 is not called when init/2 rejected the connection, since no state exists yet.

Example

defmodule MyApp.RTMPHandler do
  use RTMP.Handler

  @impl true
  def handle_publish(_app, stream_key, state) do
    if MyApp.Streams.valid_key?(stream_key),
      do: {:ok, Map.put(state, :key, stream_key)},
      else: {:reject, :invalid_key, state}
  end

  @impl true
  def handle_frame(%RTMP.Frame{} = frame, state) do
    MyApp.Pipeline.push(state.key, frame)
    {:ok, state}
  end
end

Summary

Types

connect_info()

The client's connect command, passed to handle_connect/2.

connection_info()

Connection details passed to init/2.

reason()

state()

Callbacks

handle_connect(connect_info, state)

Called on the client's connect command.

handle_disconnect(reason, state)

Called once when the connection ends. The return value is ignored.

handle_frame(t, state)

Called for every audio and video frame of the published stream, in arrival order. Return {:disconnect, reason, state} to drop the publisher.

handle_metadata(metadata, state)

Called with the stream's onMetaData, if the client sends one. Return {:disconnect, reason, state} to drop the publisher.

handle_publish(app, stream_key, state)

Called on the client's publish command. This is where stream keys are checked. stream_key is the publishing name exactly as sent, including any query string ("key?token=...").

init(connection_info, opts)

Called when a TCP connection is accepted, before the RTMP handshake. opts is the :handler_options value given to RTMP.Listener.

Types

connect_info()

@type connect_info() :: %{
  app: binary(),
  tc_url: binary() | nil,
  flash_ver: binary() | nil,
  swf_url: binary() | nil,
  page_url: binary() | nil,
  object_encoding: number(),
  properties: map(),
  remote_address: :inet.ip_address(),
  remote_port: :inet.port_number()
}

The client's connect command, passed to handle_connect/2.

:app is the application name from the RTMP URL (rtmp://host/<app>/<stream_key>). :properties is the complete command object exactly as the client sent it.

connection_info()

@type connection_info() :: %{
  remote_address: :inet.ip_address(),
  remote_port: :inet.port_number()
}

Connection details passed to init/2.

reason()

@type reason() :: term()

state()

@type state() :: term()

Callbacks

handle_connect(connect_info, state)

(optional)

@callback handle_connect(connect_info(), state()) ::
  {:ok, state()} | {:reject, reason(), state()}

Called on the client's connect command.

Default: {:ok, state}.

handle_disconnect(reason, state)

(optional)

@callback handle_disconnect(reason(), state()) :: term()

Called once when the connection ends. The return value is ignored.

Default: does nothing.

handle_frame(t, state)

@callback handle_frame(RTMP.Frame.t(), state()) ::
  {:ok, state()} | {:disconnect, reason(), state()}

Called for every audio and video frame of the published stream, in arrival order. Return {:disconnect, reason, state} to drop the publisher.

handle_metadata(metadata, state)

(optional)

@callback handle_metadata(metadata :: map(), state()) ::
  {:ok, state()} | {:disconnect, reason(), state()}

Called with the stream's onMetaData, if the client sends one. Return {:disconnect, reason, state} to drop the publisher.

Default: {:ok, state}.

handle_publish(app, stream_key, state)

@callback handle_publish(app :: binary(), stream_key :: binary(), state()) ::
  {:ok, state()} | {:reject, reason(), state()}

Called on the client's publish command. This is where stream keys are checked. stream_key is the publishing name exactly as sent, including any query string ("key?token=...").

On {:reject, reason, state} the client is told the publish failed and the connection is closed. reason is never sent to the client.

init(connection_info, opts)

(optional)

@callback init(connection_info(), opts :: term()) :: {:ok, state()} | {:reject, reason()}

Called when a TCP connection is accepted, before the RTMP handshake. opts is the :handler_options value given to RTMP.Listener.

Return {:reject, reason} to close the connection immediately — for example, to refuse an IP address.

Default: {:ok, opts}.