From d52eece03ab744fa7c18bbdf70e20b5bd9589f13 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Mon, 14 Sep 2020 21:50:59 +0300
Subject: [#2497] Prototype of video streaming for media preview proxy (WIP).

---
 lib/pleroma/helpers/media_helper.ex | 58 +++++++++++++++++++++++++++++++++++--
 1 file changed, 56 insertions(+), 2 deletions(-)

diff --git a/lib/pleroma/helpers/media_helper.ex b/lib/pleroma/helpers/media_helper.ex
index d834b4a07..0183a8625 100644
--- a/lib/pleroma/helpers/media_helper.ex
+++ b/lib/pleroma/helpers/media_helper.ex
@@ -8,6 +8,7 @@ defmodule Pleroma.Helpers.MediaHelper do
   """
 
   alias Pleroma.HTTP
+  alias Pleroma.ReverseProxy
 
   @tmp_base "/tmp/pleroma-media_preview-pipe"
 
@@ -62,7 +63,6 @@ defmodule Pleroma.Helpers.MediaHelper do
 
   def video_framegrab(url) do
     with executable when is_binary(executable) <- System.find_executable("ffmpeg"),
-         {:ok, env} <- HTTP.get(url, [], pool: :media),
          {:ok, fifo_path} <- mkfifo(),
          args = [
            "-y",
@@ -76,13 +76,66 @@ defmodule Pleroma.Helpers.MediaHelper do
            "error",
            "-"
          ] do
-      run_fifo(fifo_path, env, executable, args)
+      stream_through_fifo(url, fifo_path, executable, args)
     else
       nil -> {:error, {:ffmpeg, :command_not_found}}
       {:error, _} = error -> error
     end
   end
 
+  defp stream_loop(http_client, fifo, ffmpeg_pid) do
+    case ReverseProxy.Client.stream_body(http_client) do
+      :done ->
+        # Edge case (ffmpeg has not started processing partial input)
+        close_http_client(http_client)
+        loop_recv(ffmpeg_pid)
+
+      {:ok, data, http_client} ->
+        # Will be true if port is alive, not busy, and has received the input
+        command_sent =
+          try do
+            Port.command(fifo, data, [:nosuspend])
+          rescue
+            _ -> false
+          end
+
+        if command_sent do
+          stream_loop(http_client, fifo, ffmpeg_pid)
+        else
+          close_http_client(http_client)
+          loop_recv(ffmpeg_pid)
+        end
+
+      {:error, error} ->
+        {:error, error}
+    end
+  end
+
+  defp close_http_client(http_client) do
+    ReverseProxy.Client.close(http_client)
+  end
+
+  defp stream_through_fifo(url, fifo_path, executable, args) do
+    ffmpeg_pid =
+      Port.open({:spawn_executable, executable}, [
+        :use_stdio,
+        :stream,
+        :exit_status,
+        :binary,
+        args: args
+      ])
+
+    fifo = Port.open(to_charlist(fifo_path), [:eof, :binary, :stream, :out])
+    Process.unlink(fifo)
+
+    with {:ok, code, _headers, client} when code in 200..299 <-
+           ReverseProxy.Client.request(:get, url, [], "", pool: :media) do
+      stream_loop(client, fifo, ffmpeg_pid)
+    end
+  after
+    File.rm(fifo_path)
+  end
+
   defp run_fifo(fifo_path, env, executable, args) do
     pid =
       Port.open({:spawn_executable, executable}, [
@@ -94,6 +147,7 @@ defmodule Pleroma.Helpers.MediaHelper do
       ])
 
     fifo = Port.open(to_charlist(fifo_path), [:eof, :binary, :stream, :out])
+
     fix = Pleroma.Helpers.QtFastStart.fix(env.body)
     true = Port.command(fifo, fix)
     :erlang.port_close(fifo)
-- 
cgit v1.2.3