Unify Server.ClientHandler and Server.ClientHandlerBehaviour

lib/membrane_rtmp_plugin/rtmp/source/source.ex

@@ -13,9 +13,8 @@ defmodule Membrane.RTMP.Source do
   use Membrane.Source
   require Membrane.Logger
   require Logger
-  alias __MODULE__.ClientHandler, as: SourceClientHandler
+  alias __MODULE__.SourceClientHandler
   alias Membrane.RTMP.Server.ClientHandler
-  alias Membrane.RTMP.Utils
 
   def_output_pad :output,
     availability: :always,
@@ -82,7 +81,7 @@ defmodule Membrane.RTMP.Source do
 
   @impl true
   def handle_setup(_ctx, %{mode: :builtin_server} = state) do
-    {use_ssl?, port, app, stream_key} = Utils.parse_url(state.url)
+    {use_ssl?, port, app, stream_key} = Membrane.RTMPServer.parse_url(state.url)
 
     parent_pid = self()
 

lib/membrane_rtmp_plugin/rtmp/source/client_handler.ex → lib/membrane_rtmp_plugin/rtmp/source/source_client_handler.ex

@@ -1,10 +1,10 @@
-defmodule Membrane.RTMP.Source.ClientHandler do
+defmodule Membrane.RTMP.Source.SourceClientHandler do
   @moduledoc """
   An implementation of `Membrane.RTMP.Server.ClienHandlerBehaviour` compatible with the
   `Membrane.RTMP.Source` element.
   """
 
-  @behaviour Membrane.RTMP.Server.ClientHandlerBehaviour
+  @behaviour Membrane.RTMP.Server.ClientHandler
 
   defstruct [:controlling_process]
 

lib/membrane_rtmp_plugin/rtmp_server.ex

@@ -15,13 +15,13 @@ defmodule Membrane.RTMP.Server do
 
   require Logger
 
-  alias Membrane.RTMP.Server.ClientHandlerBehaviour
+  alias Membrane.RTMP.Server.ClientHandler
 
   @typedoc """
   Defines options for the RTMP server.
   """
   @type t :: [
-          handler: ClientHandlerBehaviour.t(),
+          handler: ClientHandler.t(),
           port: :inet.port_number(),
           use_ssl?: boolean(),
           name: atom() | nil,
@@ -97,4 +97,30 @@ defmodule Membrane.RTMP.Server do
     Enum.each(state.to_reply, &GenServer.reply(&1, port))
     {:noreply, %{state | port: port, to_reply: []}}
   end
+
+  @doc """
+  Extracts ssl, port, app and stream_key from url.
+  """
+  @spec parse_url(url :: String.t()) :: {boolean(), integer(), String.t(), String.t()}
+  def parse_url(url) do
+    uri = URI.parse(url)
+    port = uri.port
+
+    {app, stream_key} =
+      case (uri.path || "")
+           |> String.trim_leading("/")
+           |> String.trim_trailing("/")
+           |> String.split("/") do
+        [app, stream_key] -> {app, stream_key}
+        [app] -> {app, ""}
+      end
+
+    use_ssl? =
+      case uri.scheme do
+        "rtmp" -> false
+        "rtmps" -> true
+      end
+
+    {use_ssl?, port, app, stream_key}
+  end
 end

lib/membrane_rtmp_plugin/rtmp_server/client_handler.ex

@@ -9,6 +9,52 @@ defmodule Membrane.RTMP.Server.ClientHandler do
   require Logger
   alias Membrane.RTMP.{Handshake, MessageHandler, MessageParser}
 
+  @typedoc """
+  Type representing the user defined state of the client handler.
+  """
+  @type t :: term()
+
+  @doc """
+  The callback invoked once the client handler is created.
+  It should return the initial state of the client handler.
+  """
+  @callback handle_init(any()) :: t()
+
+  @doc """
+  The callback invoked when the client sends the `Membrane.RTMP.Messages.Connect.t()`
+  message.
+  """
+  @callback handle_connected(connected_msg :: Membrane.RTMP.Messages.Connect.t(), state :: t()) ::
+              t()
+
+  @doc """
+  The callback invoked when the client sends the `Membrane.RTMP.Messages.Publish.t()`
+  message.
+  """
+  @callback handle_stream_published(
+              publish_msg :: Membrane.RTMP.Messages.Publish.t(),
+              state :: t()
+            ) :: t()
+
+  @doc """
+  The callback invoked when new piece of data is received from a given client.
+  """
+  @callback handle_data_available(payload :: binary(), state :: t()) :: t()
+
+  @doc """
+  The callback invoked when the client served by given client handler
+  stops sending data.
+  (for instance, when the remote client deletes the stream or
+  terminates the socket connection)
+  """
+  @callback handle_end_of_stream(state :: t()) :: t()
+
+  @doc """
+  The callback invoked when the client handler receives a message
+  that is not recognized as an internal message of the client handler.
+  """
+  @callback handle_info(msg :: term(), t()) :: t()
+
   @doc """
   Makes the client handler ask client for the desired number of buffers
   """