Merge pull request #57 from membraneframework/fix-bugs

Fix bugs in related to browser connecting via WebRTC

Author: FelonEkonom

.circleci/config.yml

@@ -2,6 +2,22 @@ version: 2.1
 orbs:
   elixir: membraneframework/elixir@1
 
+jobs: 
+  test_exclude_browser:
+    docker:
+      - image: membraneframeworklabs/docker_membrane
+        environment:
+          MIX_ENV: test
+
+    working_directory: ~/app
+    steps:
+      - attach_workspace:
+          at: .
+      - checkout
+      - run: mix deps.get
+      - run: mix compile
+      - run: mix test --exclude browser
+
 workflows:
   version: 2
   build:
@@ -10,16 +26,16 @@ workflows:
           filters: &filters
             tags:
               only: /v.*/
-      - elixir/test:
+      - elixir/lint:
           filters:
             <<: *filters
-      - elixir/lint:
+      - test_exclude_browser:
           filters:
             <<: *filters
       - elixir/hex_publish:
           requires:
             - elixir/build_test
-            - elixir/test
+            - test_exclude_browser
             - elixir/lint
           context:
             - Deployment

README.md

@@ -70,6 +70,7 @@ For more examples, see [examples.livemd](examples.livemd).
 |---|---|---|
 | MP4 | `"*.mp4"` | `"*.mp4"` |
 | WebRTC | `{:webrtc, signaling}` | `{:webrtc, signaling}` |
+| WHIP | `{:whip, "http://*", token: "token"}` | `{:whip, "http://*", token: "token"}` |
 | RTMP | `"rtmp://*"` | _not supported_ |
 | RTSP | `"rtsp://*"` | _not supported_ |
 | RTP | `{:rtp, opts}` | _not yet supported_ |
@@ -112,7 +113,7 @@ The CLI API is a direct mapping of the Elixir API:
 For example:
 
 ```elixir
-Boombox.run(input: "file.mp4", output: {:webrtc, "ws://localhost:8830"})
+Boombox.run(input: "file.mp4", output: {:whip, "http://localhost:3721", token: "token"})
 Boombox.run(
   input:
     {:rtp,
@@ -127,7 +128,7 @@ Boombox.run(
 are equivalent to:
 
 ```sh
-./boombox -i file.mp4 -o --webrtc ws://localhost:8830
+./boombox -i file.mp4 -o --whip http://localhost:3721 --token token
 ./boombox -i --rtp --port 50001 --audio-encoding AAC --audio-specific-config a13f --aac-bitrate-mode hbr -o index.m3u8
 ```
 

bin/boombox_local

@@ -0,0 +1,2 @@
+#/bin/sh
+mix run -e 'Logger.configure(level: :info);Boombox.run_cli()' -- $@

boombox_examples_data/webrtc_from_browser.html

@@ -32,6 +32,7 @@ <h1>Boombox stream WebRTC from browser example</h1>
       const localStream = await navigator.mediaDevices.getUserMedia(mediaConstraints);
       preview.srcObject = localStream;
       const pc = new RTCPeerConnection(pcConfig);
+      window.pc = pc; // for debugging purposes
 
       pc.onicecandidate = event => {
         if (event.candidate === null) return;

boombox_examples_data/webrtc_to_browser.html

@@ -15,6 +15,7 @@ <h1>Boombox stream WebRTC to browser example</h1>
     <div>
       Boombox URL: <input type="text" value="ws://localhost:8830" id="url" /> <button id="button">Connect</button>
     </div>
+    <div id="status"></div>
     <br>
     <video id="videoPlayer" controls muted autoplay></video>
   </main>
@@ -29,6 +30,7 @@ <h1>Boombox stream WebRTC to browser example</h1>
       videoPlayer.srcObject = new MediaStream();
 
       const pc = new RTCPeerConnection(pcConfig);
+      window.pc = pc; // for debugging purposes
       pc.ontrack = event => videoPlayer.srcObject.addTrack(event.track);
       videoPlayer.play();
       pc.onicecandidate = event => {
@@ -38,6 +40,13 @@ <h1>Boombox stream WebRTC to browser example</h1>
         ws.send(JSON.stringify({ type: "ice_candidate", data: event.candidate }));
       };
 
+      pc.onconnectionstatechange = () => {
+        if (pc.connectionState == "connected") {
+          connStatus.innerHTML = "Connected";
+          button.innerHTML = "Disconnect";
+        }
+      }
+
       ws.onmessage = async event => {
         const { type, data } = JSON.parse(event.data);
 
@@ -60,7 +69,11 @@ <h1>Boombox stream WebRTC to browser example</h1>
     const connect = () => {
       const ws = new WebSocket(url.value);
       ws.onopen = () => connectRTC(ws);
-      ws.onclose = event => console.log("WebSocket connection was terminated:", event);
+      ws.onclose = event => {
+        console.log("WebSocket connection was terminated:", event);
+        connStatus.innerHTML = "Disconnected";
+        button.innerHTML = "Connect";
+      }
     }
 
     button.onclick = connect;

boombox_examples_data/whip.html

@@ -0,0 +1,56 @@
+<!DOCTYPE html>
+<html lang="en">
+
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <meta http-equiv="X-UA-Compatible" content="ie=edge">
+    <title>Membrane WebRTC WHIP/WHEP Example</title>
+</head>
+
+<body
+    style="background-color: black; color: white; font-family: Arial, Helvetica, sans-serif; min-height: 100vh; margin: 0px; padding: 5px 0px 5px 0px">
+    <h1>Boombox WHIP Example</h1>
+    <div>
+        Boombox URL: <input type="text" value="http://localhost:8829" id="url" />
+        Token: <input type="text" value="whip_it!" id="token" />
+        <button id="button">Connect</button>
+    </div>
+    <div id="status"></div>
+    <br>
+    <video id="preview" autoplay muted></video>
+    <script type="module">
+        import { WHIPClient } from 'https://cdn.jsdelivr.net/npm/[email protected]/whip.js'
+
+        const button = document.getElementById("button");
+        const connStatus = document.getElementById("status");
+        const preview = document.getElementById("preview");
+        const url = document.getElementById("url");
+        const status = document.getElementById("status");
+        const token = document.getElementById("token");
+        const pcConfig = { iceServers: [{ urls: "stun:stun.l.google.com:19302" }] };
+        const mediaConstraints = { video: true, audio: true };
+
+        const connect = async () => {
+            connStatus.innerHTML = "Connecting..."
+            const localStream = await navigator.mediaDevices.getUserMedia(mediaConstraints);
+            preview.srcObject = localStream;
+            const pc = new RTCPeerConnection(pcConfig);
+            window.pc = pc; // for debugging purposes
+            for (const track of localStream.getTracks()) { pc.addTransceiver(track, { 'direction': 'sendonly' }) }
+            const whip = new WHIPClient();
+            await whip.publish(pc, url.value, token.value);
+            connStatus.innerHTML = "Connected";
+            button.innerHTML = "Disconnect";
+            button.onclick = async () => {
+                await whip.stop();
+                status.innerHTML = "Disconnected";
+                button.onclick = connect;
+            }
+        }
+
+        button.onclick = connect;
+    </script>
+</body>
+
+</html>

examples.livemd

@@ -9,7 +9,15 @@ System.put_env("PATH", "/opt/homebrew/bin:#{System.get_env("PATH")}")
 # In case of problems installing Nx/EXLA/Bumblebee,
 # you can remove them and the Nx backend config below.
 # Examples that don't mention them should still work.
-Mix.install([:boombox, :kino, :nx, :exla, :bumblebee, :websockex, :membrane_simple_rtsp_server])
+Mix.install([
+  {:boombox, path: __DIR__},
+  :kino,
+  :nx,
+  :exla,
+  :bumblebee,
+  :websockex,
+  :membrane_simple_rtsp_server
+])
 
 Nx.global_default_backend(EXLA.Backend)
 ```
@@ -27,8 +35,7 @@ File.mkdir_p!(input_dir)
 out_dir = "#{data_dir}/output"
 File.mkdir_p!(out_dir)
 
-samples_url =
-  "https://raw.githubusercontent.com/membraneframework/static/gh-pages/samples"
+samples_url = "https://raw.githubusercontent.com/membraneframework/static/gh-pages/samples"
 
 unless File.exists?("#{input_dir}/bun.mp4") do
   %{status: 200, body: bbb_mp4} = Req.get!("#{samples_url}/big-buck-bunny/bun33s.mp4")
@@ -77,14 +84,14 @@ Boombox.run(input: {:webrtc, "ws://localhost:8829"}, output: {:webrtc, "ws://loc
 
 <!-- livebook:{"branch_parent_index":0} -->
 
-## Record WebRTC to MP4
+## Record WebRTC via WHIP to MP4
 
-To send the stream, visit http://localhost:1234/webrtc_from_browser.html.
+To send the stream, visit http://localhost:1234/whip.html.
 
 Note: don't stop this cell to finish recording - click 'disconnect' or close the browser tab instead, so the recording is finalized properly.
 
 ```elixir
-Boombox.run(input: {:webrtc, "ws://localhost:8829"}, output: "#{out_dir}/webrtc_to_mp4.mp4")
+Boombox.run(input: {:whip, "http://localhost:8829", token: "whip_it!"}, output: "#{out_dir}/webrtc_to_mp4.mp4")
 ```
 
 ```elixir
@@ -539,6 +546,7 @@ To receive the stream, visit http://localhost:1234/hls.html after running the ce
 
 ```elixir
 rtp_port = 50001
+
 t =
   Task.async(fn ->
     Boombox.run(

lib/boombox.ex

@@ -69,6 +69,7 @@ defmodule Boombox do
           (path_or_uri :: String.t())
           | {:mp4, location :: String.t(), transport: :file | :http}
           | {:webrtc, webrtc_signaling()}
+          | {:whip, uri :: String.t(), token: String.t()}
           | {:rtmp, (uri :: String.t()) | (client_handler :: pid)}
           | {:rtsp, url :: String.t()}
           | {:rtp, in_rtp_opts()}
@@ -78,6 +79,7 @@ defmodule Boombox do
           (path_or_uri :: String.t())
           | {:mp4, location :: String.t()}
           | {:webrtc, webrtc_signaling()}
+          | {:whip, uri :: String.t(), [{:token, String.t()} | {bandit_option :: atom(), term()}]}
           | {:hls, location :: String.t()}
           | {:rtp, out_rtp_opts()}
           | {:stream, out_stream_opts()}
@@ -190,6 +192,14 @@ defmodule Boombox do
       {:webrtc, uri} when is_binary(uri) ->
         value
 
+      {:whip, uri} when is_binary(uri) ->
+        parse_opt!(direction, {:whip, uri, []})
+
+      {:whip, uri, opts} when is_binary(uri) and is_list(opts) ->
+        if Keyword.keyword?(opts) do
+          {:webrtc, {:whip, uri, opts}}
+        end
+
       {:rtmp, arg} when direction == :input and (is_binary(arg) or is_pid(arg)) ->
         value
 

lib/boombox/pipeline.ex

@@ -305,8 +305,8 @@ defmodule Boombox.Pipeline do
 
   @spec create_input(Boombox.input(), Membrane.Pipeline.CallbackContext.t(), State.t()) ::
           Ready.t() | Wait.t()
-  defp create_input({:webrtc, signaling}, _ctx, state) do
-    Boombox.WebRTC.create_input(signaling, state.output, state)
+  defp create_input({:webrtc, signaling}, ctx, state) do
+    Boombox.WebRTC.create_input(signaling, state.output, ctx, state)
   end
 
   defp create_input({:mp4, location, opts}, _ctx, _state) do
@@ -331,8 +331,8 @@ defmodule Boombox.Pipeline do
 
   @spec create_output(Boombox.output(), Membrane.Pipeline.CallbackContext.t(), State.t()) ::
           {Ready.t() | Wait.t(), State.t()}
-  defp create_output({:webrtc, signaling}, _ctx, state) do
-    Boombox.WebRTC.create_output(signaling, state)
+  defp create_output({:webrtc, signaling}, ctx, state) do
+    Boombox.WebRTC.create_output(signaling, ctx, state)
   end
 
   defp create_output(_output, _ctx, state) do

lib/boombox/transcoder/video.ex

@@ -3,12 +3,16 @@ defmodule Boombox.Transcoder.Video do
 
   import Membrane.ChildrenSpec
   require Membrane.Logger
+  alias Membrane.RemoteStream
   alias Membrane.{ChildrenSpec, H264, H265, RawVideo, VP8}
 
   @type video_stream_format :: VP8.t() | H264.t() | H265.t() | RawVideo.t()
 
   defguard is_video_format(format)
-           when is_struct(format) and format.__struct__ in [VP8, H264, H265, RawVideo]
+           when is_struct(format) and
+                  (format.__struct__ in [VP8, H264, H265, RawVideo] or
+                     (format.__struct__ == RemoteStream and format.content_format == VP8 and
+                        format.type == :packetized))
 
   @spec plug_video_transcoding(
           ChildrenSpec.builder(),
@@ -70,6 +74,13 @@ defmodule Boombox.Transcoder.Video do
     builder |> child(:vp8_decoder, %VP8.Decoder{})
   end
 
+  defp maybe_plug_parser_and_decoder(builder, %RemoteStream{
+         content_format: VP8,
+         type: :packetized
+       }) do
+    builder |> child(:vp8_decoder, %VP8.Decoder{})
+  end
+
   defp maybe_plug_parser_and_decoder(builder, %RawVideo{}) do
     builder
   end

lib/boombox/utils/cli.ex

@@ -22,7 +22,9 @@ defmodule Boombox.Utils.CLI do
     audio_specific_config: {:string, :binary},
     pps: {:string, :binary},
     sps: {:string, :binary},
-    vps: {:string, :binary}
+    vps: {:string, :binary},
+    whip: {:string, :string},
+    token: {:string, :string}
   ]
 
   @spec parse_argv([String.t()]) ::