Add general use RTMP server

README.md

@@ -41,9 +41,14 @@ sudo apt-get install ffmpeg
 pacman -S ffmpeg
 ```
 
+## RTMP Server
+An simple RTMP server that accepts clients connecting on a given port and allows to distinguish between them
+based on app ID and stream key. Each client that has connected is asigned a dedicated client handler, which
+behaviour can be provided by RTMP server user.
+
 ## SourceBin
 
-Requires a socket, which has been connected to the client. It receives RTMP stream, demuxes it and outputs H264 video and AAC audio.
+Requires a client handler, which has been connected to the client, or an URL on which the client is supposed to connect. It receives RTMP stream, demuxes it and outputs H264 video and AAC audio.
 
 ## Client
 
@@ -53,10 +58,6 @@ Currently only the following codecs are supported:
 - H264 for video
 - AAC for audio
 
-## TCP Server
-
-It's a simple implementation of tcp server. It opens a tcp port and listens for incoming connections. For each new connection, a user-provided function is executed.
-
 ### Prerequisites
 
 In order to successfully build and install the plugin, you need to have **ffmpeg == 4.4** installed on your system
@@ -65,16 +66,32 @@ In order to successfully build and install the plugin, you need to have **ffmpeg
 
 ### RTMP receiver
 
-Server-side example, in which Membrane will act as an RTMP server and receive the stream, can be found under [`examples/source.exs`](examples/source.exs). Run it with:
+Server-side example, in which Membrane element will act as an RTMP server and receive the stream, can be found under [`examples/source.exs`](examples/source.exs). Run it with:
+
+```bash
+mix run examples/source.exs
+```
+
+When the server is ready you can connect to it with RTMP. If you just want to test it, you can use FFmpeg:
+
+```bash
+ffmpeg -re -i test/fixtures/testsrc.flv -f flv -c:v copy -c:a copy rtmp://localhost:1935/app/stream_key
+```
+
+### RTMP receive with standalone RTMP server
+
+If you want to see how you could setup the `Membrane.RTMP.Server` on your own and use it
+with cooperation with the `Membane.RTMP.SourceBin`, take a look at [`examples/source_with_standalone_server.exs`](examples/source_with_standalone_server.exs)
+Run it with:
 
 ```bash
-elixir examples/source.exs
+mix run examples/source.exs
 ```
 
 When the server is ready you can connect to it with RTMP. If you just want to test it, you can use FFmpeg:
 
 ```bash
-ffmpeg -re -i test/fixtures/testsrc.flv -f flv -c:v copy -c:a copy rtmp://localhost:5000
+ffmpeg -re -i test/fixtures/testsrc.flv -f flv -c:v copy -c:a copy rtmp://localhost:1935/app/stream_key
 ```
 
 ### Streaming with RTMP

examples/source.exs

@@ -1,25 +1,17 @@
-# After running this script, you can access the server at rtmp://localhost:5000
+# After running this script, you can access the server at rtmp://localhost:1935
 # You can use FFmpeg to stream to it
-# ffmpeg -re -i test/fixtures/testsrc.flv -f flv -c:v copy -c:a copy rtmp://localhost:5000
-
-Mix.install([
-  :membrane_aac_plugin,
-  :membrane_h264_plugin,
-  :membrane_flv_plugin,
-  :membrane_file_plugin,
-  {:membrane_rtmp_plugin, path: __DIR__ |> Path.join("..") |> Path.expand()}
-])
+# ffmpeg -re -i test/fixtures/testsrc.flv -f flv -c:v copy -c:a copy rtmp://localhost:1935/app/stream_key
 
 defmodule Pipeline do
   use Membrane.Pipeline
 
   @output_file "received.flv"
 
   @impl true
-  def handle_init(_ctx, socket: socket) do
+  def handle_init(_ctx, opts) do
     structure = [
       child(:source, %Membrane.RTMP.SourceBin{
-        socket: socket
+        url: "rtmp://127.0.0.1:1935/app/stream_key"
       })
       |> via_out(:audio)
       |> child(:audio_parser, %Membrane.AAC.Parser{
@@ -38,43 +30,14 @@ defmodule Pipeline do
       |> get_child(:muxer)
     ]
 
-    {[spec: structure], %{}}
-  end
-
-  # Once the source initializes, we grant it the control over the tcp socket
-  @impl true
-  def handle_child_notification(
-        {:socket_control_needed, _socket, _source} = notification,
-        :source,
-        _ctx,
-        state
-      ) do
-    send(self(), notification)
-
-    {[], state}
-  end
-
-  def handle_child_notification(_notification, _child, _ctx, state) do
-    {[], state}
-  end
-
-  @impl true
-  def handle_info({:socket_control_needed, socket, source} = notification, _ctx, state) do
-    case Membrane.RTMP.SourceBin.pass_control(socket, source) do
-      :ok ->
-        :ok
-
-      {:error, :not_owner} ->
-        Process.send_after(self(), notification, 200)
-    end
-
-    {[], state}
+    {[spec: structure], %{controller_pid: opts[:controller_pid]}}
   end
 
   # The rest of the module is used for self-termination of the pipeline after processing finishes
   @impl true
   def handle_element_end_of_stream(:sink, _pad, _ctx, state) do
-    {[terminate: :shutdown], state}
+    send(state.controller_pid, :eos)
+    {[], state}
   end
 
   @impl true
@@ -83,44 +46,15 @@ defmodule Pipeline do
   end
 end
 
-defmodule Example do
-  @server_ip {127, 0, 0, 1}
-  @server_port 5000
-
-  def run() do
-    parent = self()
+# Start a pipeline with `Membrane.RTMP.Source` that will spawn an RTMP server waiting for
+# the client connection on given URL
+{:ok, _supervisor, pipeline} = Membrane.Pipeline.start_link(Pipeline, controller_pid: self())
 
-    server_options = %Membrane.RTMP.Source.TcpServer{
-      port: @server_port,
-      listen_options: [
-        :binary,
-        packet: :raw,
-        active: false,
-        ip: @server_ip
-      ],
-      socket_handler: fn socket ->
-        # On new connection a pipeline is started
-        {:ok, _supervisor, pipeline} = Membrane.Pipeline.start_link(Pipeline, socket: socket)
-        send(parent, {:pipeline_spawned, pipeline})
-        {:ok, pipeline}
-      end
-    }
-
-    Membrane.RTMP.Source.TcpServer.start_link(server_options)
-
-    pipeline =
-      receive do
-        {:pipeline_spawned, pid} ->
-          pid
-      end
-
-    ref = Process.monitor(pipeline)
-
-    receive do
-      {:DOWN, ^ref, :process, _obj, _reason} ->
-        :ok
-    end
+# Wait for end of stream
+:ok =
+  receive do
+    :eos -> :ok
   end
-end
 
-Example.run()
+# Terminate the pipeline
+:ok = Membrane.Pipeline.terminate(pipeline)

examples/source_with_standalone_server.exs

@@ -0,0 +1,83 @@
+# After running this script, you can access the server at rtmp://localhost:1935
+# You can use FFmpeg to stream to it
+# ffmpeg -re -i test/fixtures/testsrc.flv -f flv -c:v copy -c:a copy rtmp://localhost:1935/app/stream_key
+
+defmodule Pipeline do
+  use Membrane.Pipeline
+
+  @output_file "received.flv"
+
+  @impl true
+  def handle_init(_ctx, opts) do
+    structure = [
+      child(:source, %Membrane.RTMP.SourceBin{
+        client_handler: opts[:client_handler]
+      })
+      |> via_out(:audio)
+      |> child(:audio_parser, %Membrane.AAC.Parser{
+        out_encapsulation: :none,
+        output_config: :audio_specific_config
+      })
+      |> via_in(Pad.ref(:audio, 0))
+      |> child(:muxer, Membrane.FLV.Muxer)
+      |> child(:sink, %Membrane.File.Sink{location: @output_file}),
+      get_child(:source)
+      |> via_out(:video)
+      |> child(:video_parser, %Membrane.H264.Parser{
+        output_stream_structure: :avc1
+      })
+      |> via_in(Pad.ref(:video, 0))
+      |> get_child(:muxer)
+    ]
+
+    {[spec: structure], %{controller_pid: opts[:controller_pid]}}
+  end
+
+  # The rest of the module is used for self-termination of the pipeline after processing finishes
+  @impl true
+  def handle_element_end_of_stream(:sink, _pad, _ctx, state) do
+    send(state.controller_pid, :eos)
+    {[terminate: :normal], state}
+  end
+
+  @impl true
+  def handle_element_end_of_stream(_child, _pad, _ctx, state) do
+    {[], state}
+  end
+end
+
+# Run the standalone server
+{:ok, server} =
+  Membrane.RTMP.Server.start_link(
+    behaviour: %Membrane.RTMP.Source.ClientHandler{controlling_process: self()},
+    port: 1935,
+    use_ssl?: false
+  )
+
+# Subscribe to receive client handler that connected to the
+# server with given app id and stream key
+:ok = Membrane.RTMP.Server.subscribe(server, "app", "stream_key")
+
+# Wait for the client handler
+client_handler =
+  receive do
+    {:client_handler, client_handler} ->
+      client_handler
+  end
+
+# Start the pipeline and provide it with the client_handler
+
+{:ok, _supervisor, pipeline} =
+  Membrane.Pipeline.start_link(Pipeline, client_handler: client_handler, controller_pid: self())
+
+# Wait for end of stream
+:ok =
+  receive do
+    :eos -> :ok
+  end
+
+# Terminate the server
+Process.exit(server, :normal)
+
+# Terminate the pipeline
+:ok = Membrane.Pipeline.terminate(pipeline)

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

@@ -1,4 +1,4 @@
-defmodule Membrane.RTMP.Source.DefaultBehaviourImplementation do
+defmodule Membrane.RTMP.Source.ClientHandler do
   @moduledoc """
   An implementation of `Membrane.RTMP.Server.ClienHandlerBehaviour` compatible with the
   `Membrane.RTMP.Source` element.

lib/membrane_rtmp_plugin/rtmp/source/source.ex

@@ -2,20 +2,40 @@ defmodule Membrane.RTMP.Source do
   @moduledoc """
   Membrane Element for receiving an RTMP stream. Acts as a RTMP Server.
   This implementation is limited to only AAC and H264 streams.
+
+  The source can be used in the following two scenarios:
+  * by providing the URL on which the client is expected to connect - note, that if the client doesn't
+  connect on this URL, the source won't complete its setup
+  * by spawning `Membrane.RTMP.Server`, subscribing for a given app and stream key on which the client
+  will connect, waiting for a client handler and passing the client handler to the `#{inspect(__MODULE__)}`.
   """
   use Membrane.Source
   require Membrane.Logger
   alias Membrane.RTMP.Server.ClientHandler
-  alias Membrane.RTMP.Source.DefaultBehaviourImplementation
+  alias Membrane.RTMP.Source.ClientHandler, as: SourceClientHandler
 
   def_output_pad :output,
     availability: :always,
     accepted_format: Membrane.RemoteStream,
     flow_control: :manual,
     demand_unit: :buffers
 
-  def_options client_handler: [spec: pid(), default: nil],
-              url: [spec: String.t(), default: nil]
+  def_options client_handler: [
+                default: nil,
+                spec: pid(),
+                description: """
+                A pid of a process acting as a client handler.
+                Can be gained with the use of `Membrane.RTMP.Server`.
+                """
+              ],
+              url: [
+                default: nil,
+                spec: String.t(),
+                description: """
+                An URL on which the client is expected to connect, for example:
+                rtmp://127.0.0.1:1935/app/stream_key
+                """
+              ]
 
   defguardp is_builtin_server(opts)
             when not is_nil(opts.url) and is_nil(opts.client_handler)
@@ -61,33 +81,12 @@ defmodule Membrane.RTMP.Source do
   def handle_setup(_ctx, %{mode: :builtin_server} = state) do
     {use_ssl?, port, app, stream_key} = parse_url(state.url)
 
-    listen_options =
-      if state.use_ssl? do
-        certfile = System.get_env("CERT_PATH")
-        keyfile = System.get_env("CERT_KEY_PATH")
-
-        [
-          :binary,
-          packet: :raw,
-          active: false,
-          certfile: certfile,
-          keyfile: keyfile
-        ]
-      else
-        [
-          :binary,
-          packet: :raw,
-          active: false
-        ]
-      end
-
     {:ok, server_pid} =
-      Membrane.RTMP.Server.start_link(%Membrane.RTMP.Server{
-        behaviour: %DefaultBehaviourImplementation{controlling_process: self()},
+      Membrane.RTMP.Server.start_link(
+        behaviour: %SourceClientHandler{controlling_process: self()},
         port: port,
-        use_ssl?: use_ssl?,
-        listen_options: listen_options
-      })
+        use_ssl?: use_ssl?
+      )
 
     state = %{state | app: app, stream_key: stream_key, server: server_pid}
     {[], state}
@@ -105,7 +104,7 @@ defmodule Membrane.RTMP.Source do
         {:output, %Membrane.RemoteStream{content_format: Membrane.FLV, type: :bytestream}}
     ]
 
-    :ok = DefaultBehaviourImplementation.request_for_data(state.client_handler)
+    :ok = SourceClientHandler.request_for_data(state.client_handler)
 
     {stream_format, state}
   end
@@ -140,7 +139,7 @@ defmodule Membrane.RTMP.Source do
         %{client_handler: client_handler, mode: :builtin_server} = state
       ) do
     :ok = ClientHandler.demand_data(client_handler, size)
-    :ok = DefaultBehaviourImplementation.request_for_data(state.client_handler)
+    :ok = SourceClientHandler.request_for_data(client_handler)
     {[], state}
   end