83 lines
2.4 KiB
Elixir
83 lines
2.4 KiB
Elixir
|
defmodule AugieWeb.CameraStream do
|
||
|
|
use GenServer
|
||
|
|
alias Phoenix.PubSub
|
||
|
|
import Plug.Conn
|
||
|
|
|
||
|
|
@moduledoc """
|
||
|
|
Subscribes to updated frames from the Camera.
|
||
|
|
|
||
|
|
Sends them over the `conn` at a maximum of around 24fps, dropping skipped
|
||
|
|
frames.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Maximum frame rate is around 24fps.
|
||
|
|
@frame_delay 40
|
||
|
|
|
||
|
|
def start_link, do: GenServer.start_link(__MODULE__, [])
|
||
|
|
|
||
|
|
@impl true
|
||
|
|
def init(_), do: {:ok, %{boundary: generate_boundary(), conn: nil, last_frame: nil, from: nil}}
|
||
|
|
|
||
|
|
@doc """
|
||
|
|
Call the camera streamer and ask it to start streaming to this `conn`.
|
||
|
|
|
||
|
|
The server doesn't actually reply to the call until the `terminate/2` callback
|
||
|
|
is called.
|
||
|
|
"""
|
||
|
|
@spec stream(GenServer.server(), Plug.Conn.t()) :: Plug.Conn.t()
|
||
|
|
def stream(server, conn), do: GenServer.call(server, {:start_stream, conn}, :infinity)
|
||
|
|
|
||
|
|
@impl true
|
||
|
|
def handle_call({:start_stream, conn}, from, %{conn: nil, boundary: boundary} = state) do
|
||
|
|
conn =
|
||
|
|
conn
|
||
|
|
|> put_resp_header("Age", "0")
|
||
|
|
|> put_resp_header("Cache-Control", "no-cache, private")
|
||
|
|
|> put_resp_header("Pragma", "no-cache")
|
||
|
|
|> put_resp_header("Content-Type", "multipart/x-mixed-replace; boundary=#{boundary}")
|
||
|
|
|> send_chunked(200)
|
||
|
|
|
||
|
|
PubSub.subscribe(Augie.PubSub, "camera")
|
||
|
|
Process.send_after(self(), :send_frame, @frame_delay)
|
||
|
|
{:noreply, %{state | from: from, conn: conn}}
|
||
|
|
end
|
||
|
|
|
||
|
|
@impl true
|
||
|
|
def handle_info({:frame, frame}, state), do: {:noreply, %{state | last_frame: frame}}
|
||
|
|
|
||
|
|
def handle_info(:send_frame, %{last_frame: nil} = state) do
|
||
|
|
Process.send_after(self(), :send_frame, @frame_delay)
|
||
|
|
|
||
|
|
{:noreply, state}
|
||
|
|
end
|
||
|
|
|
||
|
|
def handle_info(
|
||
|
|
:send_frame,
|
||
|
|
%{last_frame: frame, conn: conn, boundary: boundary} = state
|
||
|
|
) do
|
||
|
|
Process.send_after(self(), :send_frame, @frame_delay)
|
||
|
|
size = byte_size(frame)
|
||
|
|
header = "------#{boundary}\r\nContent-Type: image/jpeg\r\nContent-length: #{size}\r\n\r\n"
|
||
|
|
footer = "\r\n"
|
||
|
|
|
||
|
|
conn =
|
||
|
|
with {:ok, conn} <- chunk(conn, header),
|
||
|
|
{:ok, conn} <- chunk(conn, frame),
|
||
|
|
{:ok, conn} <- chunk(conn, footer),
|
||
|
|
do: conn
|
||
|
|
|
||
|
|
Process.send_after(self(), :send_frame, 40)
|
||
|
|
|
||
|
|
{:noreply, %{state | conn: conn, last_frame: nil}}
|
||
|
|
end
|
||
|
|
|
||
|
|
@impl true
|
||
|
|
def terminate(_reason, %{conn: conn, from: from}) do
|
||
|
|
GenServer.reply(from, conn)
|
||
|
|
end
|
||
|
|
|
||
|
|
defp generate_boundary do
|
||
|
|
:crypto.strong_rand_bytes(16) |> Base.encode64() |> binary_part(0, 16)
|
||
|
|
end
|
||
|
|
end
|