Add demos

Author: Noarkhh

examples.livemd

@@ -24,7 +24,8 @@ Mix.install([
   :exla,
   :bumblebee,
   :websockex,
-  :membrane_simple_rtsp_server
+  :membrane_simple_rtsp_server,
+  {:coerce, github: "ypconstante/elixir-coerce", override: true}
 ])
 
 Nx.global_default_backend(EXLA.Backend)
@@ -60,6 +61,11 @@ unless File.exists?("#{input_dir}/ffmpeg-testsrc.h264") do
   File.write!("#{input_dir}/ffmpeg-testsrc.h264", testsrc_h264)
 end
 
+unless File.exists?("#{input_dir}/test-audio.aac") do
+  %{status: 200, body: testsrc_mp4} = Req.get!("#{samples_url}/ffmpeg-testsrc-480x270.mp4")
+  File.write!("#{input_dir}/ffmpeg-testsrc.mp4", testsrc_mp4)
+end
+
 unless File.exists?("#{input_dir}/test-audio.aac") do
   %{status: 200, body: test_audio} = Req.get!("#{samples_url}/test-audio.aac")
   File.write!("#{input_dir}/test-audio.aac", test_audio)
@@ -548,6 +554,133 @@ end)
 
 <!-- livebook:{"branch_parent_index":0} -->
 
+## Compose two streams side by side, broadcast via HLS
+
+To receive the stream, visit http://localhost:1234/hls.html after running the cells below
+
+The first cell uses `:reader` and `:writer` endpoints to communicate with boombox. In this
+configuration the process calling `Boombox.read/1` controls when packets are being provided.
+
+```elixir
+input1 = "#{input_dir}/bun.mp4"
+input2 = "#{input_dir}/ffmpeg-testsrc.mp4"
+output = "#{out_dir}/index.m3u8"
+
+reader1 =
+  Boombox.run(input: input1, output: {:reader, video: :image, audio: false})
+
+reader2 =
+  Boombox.run(input: input2, output: {:reader, video: :image, audio: false})
+
+writer = Boombox.run(input: {:writer, video: :image, audio: false}, output: output)
+
+Stream.unfold(%{}, fn _state ->
+  {result1, packet1} = Boombox.read(reader1)
+  {result2, packet2} = Boombox.read(reader2)
+
+  joined_image =
+    Vix.Vips.Operation.join!(packet1.payload, packet2.payload, :VIPS_DIRECTION_HORIZONTAL)
+
+  packet = %Boombox.Packet{
+    pts: max(packet1.pts, packet2.pts),
+    payload: joined_image,
+    kind: :video
+  }
+
+  Boombox.write(writer, packet)
+
+  if :finished in [result1, result2] do
+    nil
+  else
+    {nil, %{}}
+  end
+end)
+|> Stream.run()
+
+Boombox.close(writer)
+```
+
+The second cell uses `:message` endpoints, meaning that the server communicates with boomboxes by
+exchanging messages. A consequence of this is that the inputting boomboxes will control the
+pace of providing the packets to the server, what can be useful in some circumstances:
+
+```elixir
+defmodule MyServer do
+  use GenServer
+
+  def start(args) do
+    GenServer.start(__MODULE__, args)
+  end
+
+  @impl true
+  def init(args) do
+    bb1 = Boombox.run(input: args.input1, output: {:message, video: :image, audio: false})
+    bb2 = Boombox.run(input: args.input2, output: {:message, video: :image, audio: false})
+    output_writer =
+      Boombox.run(input: {:writer, video: :image, audio: false}, output: args.output)
+
+    {:ok,
+     %{
+       input_boomboxes_states: %{
+         bb1: %{last_packet: nil, eos: false},
+         bb2: %{last_packet: nil, eos: false}
+       },
+       input_boomboxes: %{bb1 => :bb1, bb2 => :bb2},
+       output_writer: output_writer
+     }}
+  end
+
+  @impl true
+  def handle_info({:boombox_packet, bb, packet}, state) do
+    boombox_id = state.input_boomboxes[bb]
+    state = put_in(state.input_boomboxes_states[boombox_id].last_packet, packet)
+
+    if Enum.all?(Map.values(state.input_boomboxes_states), &(&1.last_packet != nil)) do
+      joined_image =
+        Vix.Vips.Operation.join!(
+          state.input_boomboxes_states.bb1.last_packet.payload,
+          state.input_boomboxes_states.bb2.last_packet.payload,
+          :VIPS_DIRECTION_HORIZONTAL
+        )
+
+      packet = %Boombox.Packet{packet | payload: joined_image}
+
+      Boombox.write(state.output_writer, packet)
+    end
+
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info({:boombox_finished, bb}, state) do
+    boombox_id = state.input_boomboxes[bb]
+    state = put_in(state.input_boomboxes_states[boombox_id].eos, true)
+
+    if Enum.all?(Map.values(state.input_boomboxes_states), & &1.eos) do
+      Boombox.close(state.output_writer)
+      {:stop, :normal, state}
+    else
+      {:noreply, state}
+    end
+  end
+end
+
+input1 = "#{input_dir}/bun.mp4"
+input2 = "#{input_dir}/ffmpeg-testsrc.mp4"
+output = "#{out_dir}/index.m3u8"
+
+{:ok, server} = MyServer.start(%{input1: input, input2: input, output: output})
+monitor = Process.monitor(server)
+
+receive do
+  {:DOWN, ^monitor, :process, ^server, reason} ->
+    IO.inspect(reason)
+    :ok
+end
+```
+
+<!-- livebook:{"branch_parent_index":0} -->
+
 ## Forward RTMP via WebRTC
 
 To receive the stream, visit http://localhost:1234/webrtc_to_browser.html

lib/boombox/server.ex

@@ -230,7 +230,7 @@ defmodule Boombox.Server do
         {:finished, packet, boombox_pid},
         %State{communication_medium: :messages, boombox_pid: boombox_pid} = state
       ) do
-    send(state.parent_pid, {:boombox_packet, packet, self()})
+    send(state.parent_pid, {:boombox_packet, self(), packet})
     send(state.parent_pid, {:boombox_finished, self()})
     {:noreply, state}
   end