wip add support for dtmf

Author: mickel8

.github/workflows/ci.yml

@@ -1,17 +1,4 @@
-name: CI
-
-on: push
-
-env:
-  MIX_ENV: test
-
-permissions:
-  contents: read
-
-jobs:
-  test:
-    runs-on: ubuntu-latest
-    name: CI on OTP ${{matrix.otp}} / Elixir ${{matrix.elixir}}
+   name: CI on OTP ${{matrix.otp}} / Elixir ${{matrix.elixir}}
     strategy:
       matrix:
         otp: ['27']
@@ -33,7 +20,6 @@ jobs:
         key: ${{ runner.os }}-mix-deps-${{ hashFiles('**/mix.lock') }}
         restore-keys: |
           ${{ runner.os }}-mix-deps-
-
     - name: Cache compiled build
       uses: actions/cache@v4
       with:
@@ -42,15 +28,13 @@ jobs:
         restore-keys: |
           ${{ runner.os }}-mix-build-
           ${{ runner.os }}-mix-
-
     - name: Cache dialyzer artifacts
       uses: actions/cache@v4
       with:
         path: _dialyzer
         key: ${{ runner.os }}-dialyzer-${{ hashFiles('**/mix.lock') }}
         restore-keys: |
           ${{ runner.os }}-dialyzer-
-
     - name: Install dependencies
       run: mix deps.get
 
@@ -84,4 +68,4 @@ jobs:
       uses: codecov/codecov-action@v4
       with:
         fail_ci_if_error: true,
-        token: ${{ secrets.CODECOV_TOKEN }}
+        token: ${{ secrets.CODECOV_TOKEN }}

examples/dtmf/.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

examples/dtmf/.gitignore

@@ -0,0 +1,23 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where third-party dependencies like ExDoc output generated docs.
+/doc/
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+dtmf-*.tar
+
+# Temporary files, for example, from tests.
+/tmp/

examples/dtmf/README.md

@@ -0,0 +1,21 @@
+# Dtmf
+
+**TODO: Add description**
+
+## Installation
+
+If [available in Hex](https://hex.pm/docs/publish), the package can be installed
+by adding `dtmf` to your list of dependencies in `mix.exs`:
+
+```elixir
+def deps do
+  [
+    {:dtmf, "~> 0.1.0"}
+  ]
+end
+```
+
+Documentation can be generated with [ExDoc](https://github.com/elixir-lang/ex_doc)
+and published on [HexDocs](https://hexdocs.pm). Once published, the docs can
+be found at <https://hexdocs.pm/dtmf>.
+

examples/dtmf/config/config.exs

@@ -0,0 +1,8 @@
+import Config
+
+config :logger, level: :info
+
+# normally you take these from env variables in `config/runtime.exs`
+config :dtmf,
+  ip: {127, 0, 0, 1},
+  port: 8829

examples/dtmf/lib/dtmf.ex

@@ -0,0 +1,15 @@
+defmodule Dtmf do
+  use Application
+
+  @ip Application.compile_env!(:dtmf, :ip)
+  @port Application.compile_env!(:dtmf, :port)
+
+  @impl true
+  def start(_type, _args) do
+    children = [
+      {Bandit, plug: __MODULE__.Router, ip: @ip, port: @port}
+    ]
+
+    Supervisor.start_link(children, strategy: :one_for_one)
+  end
+end

examples/dtmf/lib/dtmf/peer_handler.ex

@@ -0,0 +1,199 @@
+defmodule Dtmf.PeerHandler do
+  require Logger
+
+  alias ExWebRTC.{
+    ICECandidate,
+    MediaStreamTrack,
+    PeerConnection,
+    RTPCodecParameters,
+    RTP.Depayloader,
+    RTP.JitterBuffer,
+    SessionDescription
+  }
+
+  @behaviour WebSock
+
+  @ice_servers [
+    %{urls: "stun:stun.l.google.com:19302"}
+  ]
+
+  @audio_codecs [
+    %RTPCodecParameters{
+      payload_type: 111,
+      mime_type: "audio/opus",
+      clock_rate: 48_000,
+      channels: 2
+    },
+    %RTPCodecParameters{
+      payload_type: 126,
+      mime_type: "audio/telephone-event",
+      clock_rate: 8000,
+      channels: 1
+    }
+  ]
+
+  @impl true
+  def init(_) do
+    {:ok, pc} =
+      PeerConnection.start_link(
+        ice_servers: @ice_servers,
+        video_codecs: [],
+        audio_codecs: @audio_codecs
+      )
+
+    state = %{
+      peer_connection: pc,
+      in_audio_track_id: nil,
+      # The flow of this example is as follows:
+      # we first feed rtp packets into jitter buffer to
+      # wait for retransmissions and fix ordering.
+      # Once ordering and gaps are fixed, we feed packets
+      # to the depayloader, which detects DTMF events.
+      # Note that depayloader takes all RTP packets (both Opus and DTMF),
+      # but ignores those that are not DTMF ones.
+      # This is to avoid demuxing packets by the user.
+      jitter_buffer: nil,
+      depayloader: nil
+    }
+
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_in({msg, [opcode: :text]}, state) do
+    msg
+    |> Jason.decode!()
+    |> handle_ws_msg(state)
+  end
+
+  @impl true
+  def handle_info({:ex_webrtc, _from, msg}, state) do
+    handle_webrtc_msg(msg, state)
+  end
+
+  @impl true
+  def handle_info(:jitter_buffer_timeout, state) do
+    state.jitter_buffer
+    |> JitterBuffer.handle_timeout()
+    |> handle_jitter_buffer_result(state)
+  end
+
+  @impl true
+  def handle_info({:EXIT, pc, reason}, %{peer_connection: pc} = state) do
+    # Bandit traps exits under the hood so our PeerConnection.start_link
+    # won't automatically bring this process down.
+    Logger.info("Peer connection process exited, reason: #{inspect(reason)}")
+    {:stop, {:shutdown, :pc_closed}, state}
+  end
+
+  @impl true
+  def terminate(reason, _state) do
+    Logger.info("WebSocket connection was terminated, reason: #{inspect(reason)}")
+  end
+
+  defp handle_ws_msg(%{"type" => "offer", "data" => data}, state) do
+    Logger.info("Received SDP offer:\n#{data["sdp"]}")
+
+    offer = SessionDescription.from_json(data)
+    :ok = PeerConnection.set_remote_description(state.peer_connection, offer)
+
+    {:ok, answer} = PeerConnection.create_answer(state.peer_connection)
+    :ok = PeerConnection.set_local_description(state.peer_connection, answer)
+
+    answer_json = SessionDescription.to_json(answer)
+
+    msg =
+      %{"type" => "answer", "data" => answer_json}
+      |> Jason.encode!()
+
+    Logger.info("Sent SDP answer:\n#{answer_json["sdp"]}")
+
+    {:push, {:text, msg}, state}
+  end
+
+  defp handle_ws_msg(%{"type" => "ice", "data" => data}, state) do
+    Logger.info("Received ICE candidate: #{data["candidate"]}")
+
+    candidate = ICECandidate.from_json(data)
+    :ok = PeerConnection.add_ice_candidate(state.peer_connection, candidate)
+    {:ok, state}
+  end
+
+  defp handle_webrtc_msg({:connection_state_change, conn_state}, state) do
+    Logger.info("Connection state changed: #{conn_state}")
+
+    if conn_state == :failed do
+      {:stop, {:shutdown, :pc_failed}, state}
+    else
+      {:ok, state}
+    end
+  end
+
+  defp handle_webrtc_msg({:ice_candidate, candidate}, state) do
+    candidate_json = ICECandidate.to_json(candidate)
+
+    msg =
+      %{"type" => "ice", "data" => candidate_json}
+      |> Jason.encode!()
+
+    Logger.info("Sent ICE candidate: #{candidate_json["candidate"]}")
+
+    {:push, {:text, msg}, state}
+  end
+
+  defp handle_webrtc_msg({:track, %MediaStreamTrack{kind: :audio, id: id}}, state) do
+    # Find dtmf codec. Its config (payload type) might have changed during negotiation.
+    tr =
+      state.peer_connection
+      |> PeerConnection.get_transceivers()
+      |> Enum.find(fn tr -> tr.receiver.track.id == id end)
+
+    codec = Enum.find(tr.codecs, fn codec -> codec.mime_type == "audio/telephone-event" end)
+
+    if codec == nil do
+      raise "DTMF for the track has not been negotiated."
+    end
+
+    jitter_buffer = JitterBuffer.new()
+    {:ok, depayloader} = Depayloader.new(codec)
+
+    state = %{
+      state
+      | in_audio_track_id: id,
+        jitter_buffer: jitter_buffer,
+        depayloader: depayloader
+    }
+
+    {:ok, state}
+  end
+
+  defp handle_webrtc_msg({:rtp, id, nil, packet}, %{in_audio_track_id: id} = state) do
+    state.jitter_buffer
+    |> JitterBuffer.insert(packet)
+    |> handle_jitter_buffer_result(state)
+  end
+
+  defp handle_webrtc_msg(_msg, state), do: {:ok, state}
+
+  defp handle_jitter_buffer_result({packets, timeout, jitter_buffer}, state) do
+    state = %{state | jitter_buffer: jitter_buffer}
+
+    if timeout != nil do
+      Process.send_after(self(), :jitter_buffer_timeout, timeout)
+    end
+
+    state =
+      Enum.reduce(packets, state, fn packet, state ->
+        case Depayloader.depayload(state.depayloader, packet) do
+          {nil, depayloader} ->
+            %{state | depayloader: depayloader}
+
+          {event, depayloader} ->
+            Logger.info("Received DTMF event: #{event.event}")
+            %{state | depayloader: depayloader}
+        end
+      end)
+
+    {:ok, state}
+  end
+end

examples/dtmf/lib/dtmf/router.ex

@@ -0,0 +1,15 @@
+defmodule Dtmf.Router do
+  use Plug.Router
+
+  plug(Plug.Static, at: "/", from: :dtmf)
+  plug(:match)
+  plug(:dispatch)
+
+  get "/ws" do
+    WebSockAdapter.upgrade(conn, Dtmf.PeerHandler, %{}, [])
+  end
+
+  match _ do
+    send_resp(conn, 404, "not found")
+  end
+end

examples/dtmf/mix.exs

@@ -0,0 +1,32 @@
+defmodule Dtmf.MixProject do
+  use Mix.Project
+
+  def project do
+    [
+      app: :dtmf,
+      version: "0.1.0",
+      elixir: "~> 1.18",
+      start_permanent: Mix.env() == :prod,
+      deps: deps()
+    ]
+  end
+
+  # Run "mix help compile.app" to learn about applications.
+  def application do
+    [
+      extra_applications: [:logger],
+      mod: {Dtmf, []}
+    ]
+  end
+
+  # Run "mix help deps" to learn about dependencies.
+  defp deps do
+    [
+      {:plug, "~> 1.15.0"},
+      {:bandit, "~> 1.2.0"},
+      {:websock_adapter, "~> 0.5.0"},
+      {:jason, "~> 1.4.0"},
+      {:ex_webrtc, path: "../../."}
+    ]
+  end
+end