membrane_hackney_plugin

defmodule Membrane.Hackney.Sink do
  @moduledoc """
  An element uploading data over HTTP(S) based on Hackney
  """
  use Membrane.Sink

  import Mockery.Macro

  alias Membrane.Buffer

  def_input_pad :input, caps: :any, demand_unit: :bytes

  def_options location: [
                type: :string,
                description: """
                The URL of a request
                """
              ],
              method: [
                type: :atom,
                spec: :post | :put | :patch,
                description: "HTTP method that will be used when making a request",
                default: :post
              ],
              headers: [
                type: :keyword,
                description:
                  "List of additional request headers in format accepted by `:hackney.request/5`",
                default: []
              ],
              hackney_opts: [
                type: :keyword,
                description:
                  "Additional options for Hackney in format accepted by `:hackney.request/5`",
                default: []
              ],
              demand_size: [
                type: :integer,
                description: "The size of the demand made after each write",
                default: 1024
              ]

  defmodule Response do
    @moduledoc """
    Struct containing HTTP response sent to pipeline via notification after the upload is finished.
    """

    @type t :: %__MODULE__{
            status: non_neg_integer(),
            headers: [{String.t(), String.t()}],
            body: String.t()
          }

    @enforce_keys [:status, :headers, :body]
    defstruct @enforce_keys
  end

  @impl true
  def handle_init(opts) do
    state = opts |> Map.from_struct() |> Map.merge(%{conn_ref: nil})
    {:ok, state}
  end

  @impl true
  def handle_prepared_to_playing(_ctx, state) do
    {:ok, conn_ref} =
      mockable(:hackney).request(
        state.method,
        state.location,
        state.headers,
        :stream,
        state.hackney_opts
      )

    {{:ok, demand: {:input, state.demand_size}}, %{state | conn_ref: conn_ref}}
  end

  @impl true
  def handle_playing_to_prepared(_ctx, state) do
    mockable(:hackney).close(state.conn_ref)

    {:ok, %{state | conn_ref: nil}}
  end

  @impl true
  def handle_write(:input, %Buffer{payload: payload}, _ctx, state) do
    mockable(:hackney).send_body(state.conn_ref, payload)
    {{:ok, demand: {:input, state.demand_size}}, state}
  end

  @impl true
  def handle_end_of_stream(:input, _ctx, %{conn_ref: conn_ref} = state) do
    {:ok, status, headers, conn_ref} = mockable(:hackney).start_response(conn_ref)
    {:ok, body} = mockable(:hackney).body(conn_ref)

    response_notification = %__MODULE__.Response{status: status, headers: headers, body: body}
    {{:ok, notify: response_notification, notify: {:end_of_stream, :input}}, state}
  end
end