memdump

Video Streaming in Elixir

22 Apr 2024

This is blog version of the Livebook. You can try interactive nootbook here

Mix.install([
  {:req, "~> 0.4.5"},
  {:ex_cmd, "~> 0.10"},
  {:plug, "~> 1.15"},
  {:kino, "~> 0.11"},
  {:bandit, "~> 1.1"}
])

Introduction

Port is standard way to run external programs in Elixir and Erlang. However, communicating with spawned external programs over stdio can lead to various issues. ExCmd and Exile libraries tries to address these issues. On top of that they provide beloved streaming capabilities. Using them we can stream input and output all the way down to host machine.

Let’s take an example to illustrate the streaming capabilities. Here, we’ll create an HTTP server to watermark a video using FFmpeg.

The server should perform the following:

Accept request with video URL and watermark text
Read input video in a demand-driven streaming fashion
Adds watermark to the video
Stream generated video with watermark without touching disk

FFmpeg

We use a sample video from wikimedia to test.

# download sample video
video_url =
  "https://upload.wikimedia.org/wikipedia/commons/transcoded/8/81/Beachfront_B-Roll-_Fireworks_%28Free_to_Use_HD_Stock_Video_Footage%29.webm/Beachfront_B-Roll-_Fireworks_%28Free_to_Use_HD_Stock_Video_Footage%29.webm.480p.vp9.webm"

tmp = System.tmp_dir!()
input_path = Path.join(tmp, "input.webm")
video = Req.get!(video_url).body
:ok = File.write!(input_path, video)

Beachfront B-Roll- Fireworks from wikimedia

We will be using the mighty ffmpeg to add watermark. You need to install it if you haven’t already. Let’s create a helper module to wrap ffmpeg.

defmodule FFmpeg do
  @moduledoc """
  Constructs and returns ffmpeg command as list of string
  """

  @doc """
  Returns ffmpeg command for adding watermark to the input video
  """
  @spec watermark(String.t(), String.t(), String.t(), map()) :: [String.t()]
  def watermark(input, text, output, text_opts \\ []) do
    # add text with white color font and transparency of 0.5
    filter_graph =
      [
        text: "'#{text}'",
        fontsize: text_opts[:fontsize] || 80,
        fontcolor: "white",
        x: text_opts[:x] || 300,
        y: text_opts[:y] || 350,
        alpha: text_opts[:alpha] || 0.5
      ]
      |> Enum.map(fn {k, v} -> "#{k}=#{v}" end)
      |> Enum.join(":")

    [
      "ffmpeg",
      "-y",
      ["-i", input],
      ["-vf", "drawtext=#{filter_graph}"],
      ~w(-codec:a copy),
      # output should be MP4
      ~w(-f mp4),
      # add flag to fix error while reading the stream
      ~w(-movflags empty_moov),
      # output location
      output
    ]
    |> List.flatten()
  end
end

The command arguments for watermark might look intimidating, but don’t panic! ffmpeg details are not important for he demonstration.

Let’s have a look at it using the sample video we have.

output_path = Path.join(tmp, "output.mp4")

["ffmpeg" | args] = cmd = FFmpeg.watermark(input_path, "Fireworks", output_path, x: 220, y: 30)

# print the constructed ffmpeg command
IO.puts(Enum.join(cmd, " "))

# generate the video, this might take a while
{"", 0} = System.cmd(System.find_executable("ffmpeg"), args)

ffmpeg -y -i /var/folders/7m/j68yhtjd6jq7nsks6xd6nz0h0000gn/T/input.webm -vf drawtext=text='Fireworks':fontsize=80:fontcolor=white:x=220:y=30:alpha=0.5 -codec:a copy -f mp4 -movflags empty_moov /var/folders/7m/j68yhtjd6jq7nsks6xd6nz0h0000gn/T/output.mp4

output.mp4

Streaming Output

Generating video using System.cmd works, but it leave us with temporary files. Now we have additional responsibility of managing file system. Another issue is that the generated video is not available for immediate consumption. We have to wait for the encoding to finish before we use. All that is to say we should get rid of System.cmd and use ExCmd to achieve true streaming.

# `pipe:0` : read input from STDIN
# `-` : write output to STDOUT
cmd = FFmpeg.watermark("pipe:0", "Fireworks", "-", x: 220, y: 30)

# ExCmd returns output as lazy stream. You can also use `Exile` instead of `ExCmd` here
output_stream = ExCmd.stream!(cmd, input: video)

Streaming Input

ExCmd also supports input streaming. Anything which implements Enumerable or Collectable protocols can be used. Req HTTP Client supports streaming response body into a Collectable, so we can simply leverage that to feed input video to ffmpeg.

output_stream =
  ExCmd.stream!(cmd,
    input: fn ex_cmd_sink ->
      # Req pushes response (video) body segments to ex_cmd in chunks
      Req.get!(video_url, into: ex_cmd_sink)
    end)

HTTP Server

With that, the only thing remaining is to set up a HTTP server. We use Plug to create a GET /watermark route. This route accepts a video URL and watermark text as query parameters and returns the generated video as a stream using chunked response.

Streaming Output

Achieving streaming response can be done using Plug.Conn.chunk/2, allowing us to push chunks as soon as they become available from the ffmpeg command.

defmodule FFmpegServer do
  @moduledoc """
  HTTP server for demonstrating FFmpeg streaming
  """
  use Plug.Router
  require Logger

  plug(Plug.Parsers, parsers: [], pass: ["*/*"])
  plug(:match)
  plug(:dispatch)

  get "/watermark" do
    %{"video_url" => video_url, "text" => text} = conn.params

    cmd = FFmpeg.watermark("pipe:0", text, "-", x: 20, y: 20)
    output_stream = ExCmd.stream!(cmd, input: &Req.get!(video_url, into: &1))

    conn =
      conn
      |> put_resp_content_type("video/mp4")
      |> send_chunked(200)

    Enum.reduce_while(output_stream, conn, fn chunk, conn ->
      case chunk(conn, chunk) do
        {:ok, conn} ->
          Logger.debug("Sent #{IO.iodata_length(chunk)} bytes")
          {:cont, conn}

        {:error, :closed} ->
          Logger.debug("Connection closed")
          {:halt, conn}
      end
    end)
  end
end

Start the server at port 8989

webserver = {Bandit, plug: FFmpegServer, scheme: :http, port: 8989}
{:ok, supervisor} = Supervisor.start_link([webserver], strategy: :one_for_one)

Let’s test it out in shell using curl. You can even directly play the video stream using ffplay!

curl \
  --get 'http://127.0.0.1:8989/watermark' \
  --data-urlencode 'text=FFmpeg Rocks!' \
  --data-urlencode 'video_url=https://upload.wikimedia.org/wikipedia/commons/transcoded/8/81/Beachfront_B-Roll-_Fireworks_%28Free_to_Use_HD_Stock_Video_Footage%29.webm/Beachfront_B-Roll-_Fireworks_%28Free_to_Use_HD_Stock_Video_Footage%29.webm.480p.vp9.webm' \
  | ffplay -

If you notice the server logs, we can see that video chunks are fetched, encoded, and pushed, simultaneously on the fly!

Another noteworthy feature is that if the connection drops midway. For example, close ffplay playback by pressing Ctrl+C, then the streaming pipeline halts gracefully, avoiding the wasteful encoding of the entire video.