FFmpeg with Scala: real-time video

Kevin van Zonneveld

Co-founder · Amsterdam, The Netherlands ·

Use Scala to start a live RTSP test feed, transform it with FFmpeg, and serve the resulting HLS stream over local HTTP. You will see a moving grayscale pattern with a timestamp and retain a 12-second recording. The example runs on Linux, needs no camera, and stops its own services when the demonstration ends.

The important boundary is process ownership: Scala starts FFmpeg and waits for its actual exit, while keeping a deadline and shutdown hook that can stop it. Waiting for a Scala Future alone does not cancel an external encoder. This is one video-only stream at one quality, with ordinary HLS buffering; it does not demonstrate a production latency target or adaptive bitrate delivery.

Setting up Scala and FFmpeg

This walkthrough was tested on Linux x86-64 with OpenJDK 21.0.12.1, Scala CLI 1.17.1, Scala 2.13.18, MediaMTX 1.21.1, and FFmpeg/ffprobe/ffplay 9.0.1. Use Bash for the commands. Install an FFmpeg build for Linux with libx264, the HLS muxer, and the testsrc2, scale, hue, drawbox, and drawtext filters. You also need that JDK, cURL, gzip, tar, awk, fontconfig’s fc-match, and an installed TrueType font such as Liberation Sans. FFplay needs a graphical desktop for visible playback.

java -version && ffmpeg -version && ffprobe -version && ffplay -version &&
  fc-match 'Liberation Sans'

Create a new project and download the pinned Scala CLI and MediaMTX releases. MediaMTX is the local RTSP counterpart: FFmpeg publishes to it, then another FFmpeg process reads that stream. Paste this block from a writable parent directory. An existing scala-live-demo makes setup stop without changing that directory; choose a different parent to start again.

(
  mkdir scala-live-demo &&
  cd scala-live-demo &&
  curl -fsSLo scala-cli.gz https://github.com/VirtusLab/scala-cli/releases/download/v1.17.1/scala-cli-x86_64-pc-linux.gz &&
  gzip -d scala-cli.gz &&
  chmod +x scala-cli &&
  curl -fsSLo mediamtx.tar.gz https://github.com/bluenviron/mediamtx/releases/download/v1.21.1/mediamtx_v1.21.1_linux_amd64.tar.gz &&
  tar -xzf mediamtx.tar.gz mediamtx &&
  font_file=$(fc-match -f '%{file}' 'Liberation Sans') &&
  test -r "$font_file" &&
  cp "$font_file" font.ttf &&
  ./scala-cli version && ./mediamtx --version
)

Stay in the parent directory for the remaining commands. Each parenthesized block keeps its directory changes and shell options local. A failed download can leave an incomplete new project; inspect it and use a new parent before retrying. Save the complete program below as scala-live-demo/Live.scala. The first build downloads the Scala compiler and libraries; there are no additional application dependencies. The commands target Live.scala explicitly; an enclosing sbt project is not the build input.

Integrating FFmpeg with Scala

The program uses Java’s ProcessBuilder with separate arguments, rather than interpolating a shell command. It writes a private MediaMTX configuration, binds HTTP to loopback, and chooses available ports by default. Only the playlist and completed segment filenames are served; configuration, fonts, and logs stay off the HTTP surface.

//> using scala 2.13.18

import com.sun.net.httpserver.{HttpExchange, HttpServer}
import java.net.{InetAddress, InetSocketAddress, ServerSocket}
import java.nio.file.{Files, LinkOption, Path, Paths}
import java.util.concurrent.TimeUnit.SECONDS
import scala.collection.mutable.ArrayBuffer
import scala.util.control.NonFatal

object Live {
  private val children = ArrayBuffer.empty[(String, java.lang.Process)]
  private var http: Option[HttpServer] = None
  private var stopping = false
  @volatile private var encoded = false

  private def stopOwned(): Unit = synchronized {
    stopping = true
    http.foreach(_.stop(0))
    http = None
    for ((name, child) <- children.reverse if child.isAlive) {
      child.destroy()
      if (!child.waitFor(2, SECONDS)) {
        child.destroyForcibly()
        require(child.waitFor(2, SECONDS), s"Could not stop $name")
      }
    }
    children.clear()
  }

  private def start(name: String, args: Seq[String], out: Path): java.lang.Process = synchronized {
    require(!stopping, "Run is stopping")
    val builder = new ProcessBuilder(args: _*)
      .directory(out.toFile)
      .redirectErrorStream(true)
      .redirectOutput(out.resolve(s"$name.log").toFile)
    // Ambient MediaMTX overrides must not reopen listeners disabled by our configuration.
    builder.environment().keySet().removeIf(key => key.startsWith("MTX_"))
    val child = builder.start()
    children += name -> child
    println(s"Started $name (PID ${child.pid()})")
    child
  }

  private def serve(exchange: HttpExchange, out: Path): Unit = {
    try {
      val name = exchange.getRequestURI.getPath.stripPrefix("/")
      val allowed = name == "stream.m3u8" || name.matches("segment-[0-9]{3}\\.ts")
      val file = out.resolve(name)
      if (exchange.getRequestMethod != "GET") exchange.sendResponseHeaders(405, -1)
      else if (!allowed || !Files.isRegularFile(file)) exchange.sendResponseHeaders(404, -1)
      else {
        val contentType = if (name.endsWith(".ts")) "video/mp2t" else "application/vnd.apple.mpegurl"
        exchange.getResponseHeaders.set("Content-Type", contentType)
        exchange.getResponseHeaders.set("Cache-Control", "no-store")
        val bytes = Files.readAllBytes(file)
        exchange.sendResponseHeaders(200, bytes.length)
        exchange.getResponseBody.write(bytes)
      }
    } finally exchange.close()
  }

  private def run(args: Array[String]): Unit = {
    require(args.length >= 2 && args.length <= 4,
      "Usage: Live <new-directory> <deadline-seconds> [rtsp-port] [http-port]")
    val seconds = args(1).toInt
    require(seconds >= 1 && seconds <= 120, "Deadline must be between 1 and 120 seconds")
    val ports = args.drop(2).map(_.toInt)
    require(ports.forall(port => port >= 0 && port <= 65535), "Invalid port")
    val project = Paths.get(".").toRealPath()
    val out = project.resolve(args(0)).normalize()
    require(out.getParent == project, "Use a new direct child directory of this project")
    require(!Files.exists(out, LinkOption.NOFOLLOW_LINKS), "Output directory already exists")
    require(Files.isReadable(project.resolve("font.ttf")), "Missing readable font.ttf")
    Files.createDirectory(out)
    Files.copy(project.resolve("font.ttf"), out.resolve("font.ttf"))
    val deadline = System.nanoTime() + SECONDS.toNanos(seconds)
    def checkDeadline(): Unit =
      require(System.nanoTime() < deadline, "Run deadline exceeded; output may be partial")
    def alive(name: String, child: java.lang.Process): Unit =
      require(child.isAlive, s"$name stopped (exit ${child.exitValue()}); inspect $name.log")
    def waitUntil(ready: => Boolean)(check: => Unit): Unit = {
      while (!ready) { checkDeadline(); check; Thread.sleep(100) }
      checkDeadline()
    }
    val hook = new Thread(() => {
      Console.err.println(if (encoded) "Stopped; completed HLS retained" else "Cancelled; output may be partial")
      stopOwned()
    })
    Runtime.getRuntime.addShutdownHook(hook)
    try {
      println(s"Controller PID ${ProcessHandle.current().pid()}")
      val requestedRtsp = ports.headOption.getOrElse(0)
      val reservation = new ServerSocket(requestedRtsp, 1, InetAddress.getByName("127.0.0.1"))
      val rtspPort = reservation.getLocalPort
      reservation.close()
      val server = synchronized {
        require(!stopping, "Run is stopping")
        val listener = HttpServer.create(new InetSocketAddress("127.0.0.1", ports.lift(1).getOrElse(0)), 0)
        http = Some(listener)
        listener.createContext("/", exchange => serve(exchange, out))
        listener.start()
        listener
      }
      val url = s"rtsp://127.0.0.1:$rtspPort/live"
      val config = out.resolve("mediamtx.yml")
      Files.writeString(config, s"""logLevel: info
rtsp: yes
rtspAddress: 127.0.0.1:$rtspPort
rtspTransports: [tcp]
rtmp: no
hls: no
webrtc: no
srt: no
moq: no
api: no
metrics: no
pprof: no
playback: no
paths:
  live:
    source: publisher
""")
      val mtx = start("mediamtx", Seq(project.resolve("mediamtx").toString, config.toString), out)
      def serverLog: String = Files.readString(out.resolve("mediamtx.log"))
      waitUntil(serverLog.contains("[RTSP] started with listeners")) { alive("mediamtx", mtx) }
      val common = Seq("ffmpeg", "-hide_banner", "-loglevel", "warning", "-nostdin")
      val source = start("source", common ++ Seq(
        "-re", "-f", "lavfi", "-i", "testsrc2=size=320x180:rate=10", "-t", "45",
        "-an", "-c:v", "libx264", "-threads", "2", "-preset", "ultrafast",
        "-tune", "zerolatency", "-g", "10", "-pix_fmt", "yuv420p",
        "-f", "rtsp", "-rtsp_transport", "tcp", url), out)
      waitUntil(serverLog.contains("is publishing to path 'live'")) {
        alive("mediamtx", mtx); alive("source", source)
      }
      val filter = "scale=160:90,hue=s=0,drawbox=x=0:y=0:w=iw:h=22:color=black:t=fill," +
        "drawtext=fontfile=font.ttf:text='Scala %{pts\\:hms}':fontcolor=white:fontsize=12:x=5:y=4"
      val encoder = start("encoder", common ++ Seq(
        "-n", "-rtsp_transport", "tcp", "-timeout", "3000000", "-i", url,
        "-map", "0:v:0", "-an", "-vf", filter, "-filter_threads", "2", "-frames:v", "120",
        "-c:v", "libx264", "-threads", "2", "-preset", "ultrafast", "-tune", "zerolatency",
        "-pix_fmt", "yuv420p", "-g", "20", "-sc_threshold", "0", "-flags", "+cgop",
        "-f", "hls", "-hls_time", "2", "-hls_playlist_type", "event",
        "-hls_flags", "independent_segments+temp_file", "-hls_segment_filename", "segment-%03d.ts",
        "-progress", "progress.txt", "stream.m3u8"), out)
      val playlist = out.resolve("stream.m3u8")
      waitUntil(Files.exists(playlist)) {
        alive("mediamtx", mtx); alive("source", source); alive("encoder", encoder)
      }
      val playback = s"http://127.0.0.1:${server.getAddress.getPort}/stream.m3u8"
      Files.writeString(out.resolve("playback.url"), playback + "\n")
      println(s"Play: $playback")
      waitUntil(!encoder.isAlive) { alive("mediamtx", mtx); alive("source", source) }
      require(encoder.exitValue() == 0, "Encoder failed; inspect encoder.log")
      val frames = Files.readString(out.resolve("progress.txt")).linesIterator
        .filter(_.startsWith("frame=")).toVector.lastOption
      require(frames.exists(_.stripPrefix("frame=").trim == "120"), "Incomplete video; output may be partial")
      require(Files.readString(playlist).contains("#EXT-X-ENDLIST"), "Playlist was not finalized")
      encoded = true
      Files.writeString(out.resolve("complete.txt"), "120 frames encoded\n")
      println("Encoding complete; HTTP remains available until the run deadline")
      while (System.nanoTime() < deadline) Thread.sleep(100)
    } finally {
      stopOwned()
      Runtime.getRuntime.removeShutdownHook(hook)
    }
    println("Completed; services stopped and HLS retained")
  }

  def main(args: Array[String]): Unit = {
    try run(args)
    catch { case NonFatal(error) => Console.err.println(s"Live demo failed: ${error.getMessage}"); sys.exit(1) }
  }
}

Real-time video stream processing

FFmpeg’s -re paces the generated source instead of publishing it as fast as the CPU can encode it. MediaMTX accepts the publisher on /live; the encoder reads that same RTSP path using TCP transport. This example uses RTSP, rather than assuming that RTMP or an arbitrary camera has the same input behavior.

The filter resizes to 160 by 90 pixels, removes color, and draws a black band with an elapsed-time timestamp. drawtext needs font support in your FFmpeg build even though the font file is explicit. Builds with fontconfig can fall back to another font if that file is invalid, so use the installed font selected during setup and check the visible overlay. A video filter requires decoding and re-encoding; -c:v copy cannot apply it.

For HLS, the encoder aligns keyframes every 20 frames at 10 fps, matching the two-second segment target. event retains all segments, and temp_file makes each segment available under its final name after writing finishes. The run produces 120 frames and then finalizes the playlist. This is a bounded recording of a live source, not an unlimited sliding-window archive.

Build an executable JAR and start the demo. --jvm system uses your installed JDK; --server=false avoids a background compilation server. Scala CLI’s --assembly packaging includes the Scala runtime. -f replaces only this example’s live.jar, and && prevents an old JAR from running when rebuilding fails.

(
  cd scala-live-demo &&
    ./scala-cli --power package Live.scala --assembly --preamble=false --server=false --jvm system -f -o live.jar &&
    java -jar live.jar hls 60
)

The 60-second run deadline starts after argument validation and font staging. It covers service startup, encoding, and the playback window; compilation and the shutdown waits are outside it. Both ports are chosen automatically. Optional third and fourth program arguments select the RTSP and HTTP ports; zero requests an available port. An occupied requested port fails startup. Each run needs a new output directory directly inside the project, so a rerun with hls preserves the existing output.

Testing FFmpeg integrations

When the program prints the playback URL, open a second terminal in the same parent directory and run this block. The HTTP server is available while encoding and until the run deadline. You should see a moving grayscale pattern with a white timestamp in the top band. -autoexit closes FFplay when the finalized stream ends; press q to close it sooner.

(
  playback_url=$(cat scala-live-demo/hls/playback.url) &&
    ffplay -autoexit "$playback_url"
)

After the demo exits successfully, scala-live-demo/hls contains stream.m3u8, six segment-000.ts through segment-005.ts files, and complete.txt. Inspect the stream and decode the full recording from its actual playlist:

(
  set -euo pipefail
  cd scala-live-demo/hls &&
  ffprobe -v error -show_entries stream=codec_name,width,height:format=duration \
    -of json stream.m3u8 &&
  ffmpeg -hide_banner -loglevel error -nostdin -xerror -progress pipe:1 -i stream.m3u8 \
    -map 0:v:0 -f null - |
    awk -F= '$1 == "frame" {frames=$2} END {if (frames != 120) {print "Expected 120 decoded frames" > "/dev/stderr"; exit 1}}'
)

Expect H.264 video at 160 by 90 pixels and a playlist duration of 12 seconds. Probing and decoding check the recording’s structure and readability; playback checks the visible filter result. The frame-count check matters because FFmpeg can skip a missing HLS segment and still return zero, even with -xerror. A zero encoder status does not establish an intact camera feed, perceptual quality, or end-to-end latency.

Error handling and resource management

The controller checks that the publisher and RTSP server remain alive while the encoder works. A disconnect can make FFmpeg finish with fewer frames, even with exit status zero; the progress check refuses to mark that recording complete. Encoder failure, a stopped source, and a deadline before encoding finishes return status one. Logs identify each child as mediamtx.log, source.log, or encoder.log. Failed encoding runs can retain partial media for inspection and have no complete.txt.

Press Ctrl+C in the demo’s first terminal to stop it early. The JVM shutdown hook blocks new child starts, stops HTTP, and terminates its child processes, waiting up to two seconds per child before forced termination and another two-second wait. This applies to ordinary interruption and termination, not an uncatchable kill or machine shutdown. Output already written remains; cancellation during encoding does not report completion. After encoding succeeds, an early stop retains the completed recording.

At the normal deadline the program stops its services and returns status zero if encoding completed. Keep the output directory to replay the local playlist, or remove that specific directory after inspection. The program does not delete or overwrite an earlier run, and the project’s downloaded tools and JAR remain available for another run with a new directory name.

Troubleshooting common issues

If startup fails, read the named log and check the executable and font prerequisites. The program stops already-started children before returning an error; it prints no playback URL before a playlist is available.
If FFplay cannot connect, start it while the demo is still running. After shutdown, use ffplay -autoexit scala-live-demo/hls/stream.m3u8 to replay completed local files.
If encoding misses the deadline, inspect encoder.log and choose a deadline up to 120 seconds with a new output directory. Increasing the deadline does not repair a disconnected publisher.

For a different filter, change the single filter value, rebuild the JAR, and launch it with a new output directory name. Keep FFmpeg’s thread limits while measuring whether encoding can keep pace with the source. Adding Scala Futures would start more encoders; it would not make this encoder faster or give those processes cancellation. For existing local recordings instead, the Scala concatenation guide covers joining compatible files without re-encoding.