FFmpeg con Scala: video en tiempo real

Kevin van Zonneveld

Cofundador · Ámsterdam, Países Bajos ·

Usa Scala para iniciar una fuente RTSP de prueba en directo, transformarla con FFmpeg y servir el flujo HLS resultante por HTTP local. Verás un patrón en movimiento en escala de grises con una marca de tiempo y conservarás una grabación de 12 segundos. El ejemplo se ejecuta en Linux, no necesita cámara y detiene sus propios servicios cuando termina la demostración.

El límite importante es la gestión de los procesos propios: Scala inicia FFmpeg y espera a que termine realmente, mientras mantiene un plazo límite y un hook de cierre que pueden detenerlo. Esperar únicamente a un Future de Scala no cancela un codificador externo. Se trata de un solo flujo de video sin audio, con una sola calidad y el almacenamiento en búfer habitual de HLS; no demuestra un objetivo de latencia de producción ni una entrega con tasa de bits adaptativa.

Configurar Scala y FFmpeg

Este tutorial se probó en Linux x86-64 con OpenJDK 21.0.12.1, Scala CLI 1.17.1, Scala 2.13.18, MediaMTX 1.21.1 y FFmpeg/ffprobe/ffplay 9.0.1. Usa Bash para los comandos. Instala una compilación de FFmpeg para Linux con libx264, el multiplexor HLS y los filtros testsrc2, scale, hue, drawbox y drawtext. También necesitas ese JDK, cURL, gzip, tar, awk, fc-match de fontconfig y una fuente TrueType instalada, como Liberation Sans. FFplay necesita un escritorio gráfico para mostrar la reproducción.

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

Crea un proyecto nuevo y descarga las versiones fijadas de Scala CLI y MediaMTX. MediaMTX es el servidor RTSP local: FFmpeg publica en él y luego otro proceso de FFmpeg lee ese flujo. Pega este bloque desde un directorio padre en el que tengas permiso de escritura. Si ya existe scala-live-demo, la configuración se detiene sin modificar ese directorio; elige otro directorio padre para volver a empezar.

(
  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
)

Permanece en el directorio padre para los comandos restantes. Cada bloque entre paréntesis mantiene sus cambios de directorio y opciones de shell dentro de su propio ámbito. Una descarga fallida puede dejar incompleto el proyecto nuevo; inspecciónalo y usa otro directorio padre antes de reintentar. Guarda el programa completo que aparece a continuación como scala-live-demo/Live.scala. La primera compilación descarga el compilador y las bibliotecas de Scala; no hay dependencias adicionales de la aplicación. Los comandos apuntan explícitamente a Live.scala; un proyecto sbt que lo contenga no se usa como entrada de compilación.

Integrar FFmpeg con Scala

El programa usa ProcessBuilder de Java con argumentos separados, en lugar de interpolar un comando de shell. Escribe una configuración privada de MediaMTX, vincula HTTP a la interfaz de loopback y elige puertos disponibles de forma predeterminada. Solo se sirven la lista de reproducción y los archivos de segmentos completados bajo sus nombres; la configuración, las fuentes y los registros no se exponen por HTTP.

//> 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) }
  }
}

Procesamiento de flujos de video en tiempo real

-re de FFmpeg regula el ritmo de la fuente generada en lugar de publicarla tan rápido como la CPU pueda codificarla. MediaMTX acepta al publicador en /live; el codificador lee esa misma ruta RTSP mediante transporte TCP. Este ejemplo usa RTSP, en lugar de suponer que RTMP o una cámara cualquiera tienen el mismo comportamiento de entrada.

El filtro redimensiona a 160 por 90 píxeles, elimina el color y dibuja una banda negra con una marca de tiempo que muestra el tiempo transcurrido. drawtext necesita soporte para fuentes en tu compilación de FFmpeg, aunque el archivo de fuente se indique explícitamente. Las compilaciones con fontconfig pueden recurrir a otra fuente si ese archivo no es válido, así que usa la fuente instalada que seleccionaste durante la configuración y comprueba la superposición visible. Un filtro de video requiere decodificación y recodificación; -c:v copy no puede aplicarlo.

Para HLS, el codificador alinea los fotogramas clave cada 20 fotogramas a 10 fps, en consonancia con el objetivo de segmentos de dos segundos. event conserva todos los segmentos y temp_file hace que cada segmento esté disponible con su nombre definitivo una vez finalizada la escritura. La ejecución produce 120 fotogramas y después finaliza la lista de reproducción. Es una grabación acotada de una fuente en directo, no un archivo ilimitado de ventana deslizante.

Compila un JAR ejecutable e inicia la demostración. --jvm system usa tu JDK instalado; --server=false evita un servidor de compilación en segundo plano. El empaquetado --assembly de Scala CLI incluye el entorno de ejecución de Scala. -f reemplaza únicamente el live.jar de este ejemplo y && impide que se ejecute un JAR antiguo cuando falla la recompilación.

(
  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
)

El plazo límite de ejecución de 60 segundos comienza después de validar los argumentos y preparar la fuente tipográfica. Abarca el inicio de los servicios, la codificación y el período de reproducción; la compilación y las esperas durante el cierre quedan fuera. Ambos puertos se eligen automáticamente. Los argumentos opcionales tercero y cuarto del programa seleccionan los puertos RTSP y HTTP; cero solicita un puerto disponible. Si el puerto solicitado está ocupado, el inicio falla. Cada ejecución necesita un directorio de salida nuevo directamente dentro del proyecto, por lo que volver a ejecutar con hls conserva la salida existente.

Probar las integraciones de FFmpeg

Cuando el programa imprima la URL de reproducción, abre una segunda terminal en el mismo directorio padre y ejecuta este bloque. El servidor HTTP está disponible durante la codificación y hasta el plazo límite de ejecución. Deberías ver un patrón en movimiento en escala de grises con una marca de tiempo blanca en la banda superior. -autoexit cierra FFplay cuando termina el flujo finalizado; presiona q para cerrarlo antes.

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

Una vez que la demostración termina correctamente, scala-live-demo/hls contiene stream.m3u8, seis archivos desde segment-000.ts hasta segment-005.ts y complete.txt. Inspecciona el flujo y decodifica la grabación completa a partir de su lista de reproducción real:

(
  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}}'
)

El resultado esperado es video H.264 de 160 por 90 píxeles y una lista de reproducción con una duración de 12 segundos. El análisis y la decodificación comprueban la estructura y la legibilidad de la grabación; la reproducción comprueba el resultado visible del filtro. La comprobación del número de fotogramas es importante porque FFmpeg puede omitir un segmento HLS faltante y aun así devolver cero, incluso con -xerror. Un estado de salida cero del codificador no demuestra que la señal de la cámara esté intacta, ni establece la calidad perceptual o la latencia de extremo a extremo.

Manejo de errores y gestión de recursos

El controlador comprueba que el publicador y el servidor RTSP sigan activos mientras trabaja el codificador. Una desconexión puede hacer que FFmpeg termine con menos fotogramas, incluso con un estado de salida cero; la comprobación del progreso impide marcar esa grabación como completa. Un fallo del codificador, una fuente detenida o el vencimiento del plazo antes de que termine la codificación devuelven el estado uno. Los registros identifican cada proceso hijo como mediamtx.log, source.log o encoder.log. Las ejecuciones de codificación fallidas pueden conservar contenido multimedia parcial para su inspección y no tienen complete.txt.

Presiona Ctrl+C en la primera terminal de la demostración para detenerla antes de tiempo. El hook de cierre de la JVM bloquea el inicio de nuevos procesos hijos, detiene HTTP y termina sus procesos hijos. Espera hasta dos segundos por proceso hijo antes de forzar su terminación y vuelve a esperar dos segundos. Esto se aplica a las señales ordinarias de interrupción y terminación, no a una terminación que no pueda interceptarse ni al apagado del equipo. La salida ya escrita permanece; la cancelación durante la codificación no informa que esta se haya completado. Una vez que la codificación termina correctamente, detener el programa antes de tiempo conserva la grabación completa.

Al cumplirse el plazo normal, el programa detiene sus servicios y devuelve el estado cero si se completó la codificación. Conserva el directorio de salida para volver a reproducir la lista local, o elimina ese directorio específico después de inspeccionarlo. El programa no elimina ni sobrescribe una ejecución anterior, y las herramientas descargadas y el JAR del proyecto siguen disponibles para otra ejecución con un nombre de directorio nuevo.

Solucionar problemas comunes

Si el inicio falla, lee el registro indicado y comprueba los requisitos de ejecutables y fuentes tipográficas. El programa detiene los procesos hijos ya iniciados antes de devolver un error; no imprime ninguna URL de reproducción hasta que haya una lista de reproducción disponible.
Si FFplay no puede conectarse, inícialo mientras la demostración siga en ejecución. Después del cierre, usa ffplay -autoexit scala-live-demo/hls/stream.m3u8 para volver a reproducir los archivos locales completos.
Si la codificación no termina dentro del plazo, inspecciona encoder.log y elige un plazo de hasta 120 segundos con un directorio de salida nuevo. Aumentar el plazo no soluciona la desconexión de un publicador.

Para usar otro filtro, cambia el único valor filter, vuelve a compilar el JAR y ejecútalo con un nombre de directorio de salida nuevo. Mantén los límites de hilos de FFmpeg mientras mides si la codificación puede seguir el ritmo de la fuente. Añadir más instancias de Future de Scala iniciaría más codificadores; no haría que este codificador fuera más rápido ni permitiría cancelar esos procesos. Para trabajar con grabaciones locales existentes, la guía de concatenación con Scala explica cómo unir archivos compatibles sin recodificarlos.