defmodule K8s.Client.Mint.Request.HTTP do @moduledoc """ Represents a HTTP request state. ### Fields - `:caller` - Synchronous requests only: The calling process. - `:stream_to` - StreamTo requests only: The process expecting response parts sent to. - `:waiting` - Streamed requests only: The process waiting for the next response part. - `:response` - The response containing received parts. """ alias K8s.Client.Mint.Request.Upgrade, as: UpgradeRequest alias K8s.Client.Mint.Request.WebSocket, as: WebSocketRequest @data_types [:data, :stdout, :stderr, :error, :waiting] @type t :: %__MODULE__{ caller: pid() | nil, stream_to: pid() | nil, waiting: pid() | nil, response: %{} } @type request :: t() | WebSocketRequest.t() | UpgradeRequest.t() defstruct [:caller, :stream_to, :waiting, response: %{}] @spec new(keyword()) :: t() def new(fields), do: struct!(__MODULE__, fields) @spec put_response(request(), :done | {atom(), term()}) :: :pop | {request(), request()} def put_response(%{response: response, caller: caller}, :done) when not is_nil(caller) do response = response |> Map.update(:data, nil, &(&1 |> Enum.reverse() |> IO.iodata_to_binary())) |> Map.reject(&(&1 |> elem(1) |> is_nil())) GenServer.reply(caller, {:ok, response}) :pop end def put_response(%{response: response, caller: caller} = request, {:close, data}) when not is_nil(caller) do response = response |> Map.update(:stdout, nil, &(&1 |> Enum.reverse() |> IO.iodata_to_binary())) |> Map.update(:stderr, nil, &(&1 |> Enum.reverse() |> IO.iodata_to_binary())) |> Map.update(:error, nil, &(&1 |> Enum.reverse() |> IO.iodata_to_binary())) |> Map.reject(&(&1 |> elem(1) |> is_nil())) |> Map.put(:close, data) GenServer.reply(caller, {:ok, response}) {:stop, request} end def put_response(%{stream_to: stream_to} = request, {:close, data}) when not is_nil(stream_to) do send(stream_to, {:close, data}) {:stop, request} end def put_response(%{stream_to: stream_to}, :done) when not is_nil(stream_to) do send(stream_to, {:done, true}) :pop end def put_response(request, {:close, data}) do {request, put_in(request.response[:close], data)} end def put_response(request, :done) do {request, put_in(request.response[:done], true)} end def put_response(%{stream_to: stream_to} = request, {type, new_data}) when type in @data_types and not is_nil(stream_to) do send(stream_to, {type, new_data}) {request, request} end def put_response(request, {type, new_data}) when type in @data_types do {request, update_in(request, [Access.key(:response), Access.key(type, [])], &[new_data | &1])} end def put_response(%{stream_to: stream_to} = request, {type, value}) when not is_nil(stream_to) do send(stream_to, {type, value}) {request, request} end def put_response(request, {type, value}) do {request, put_in(request.response[type], value)} end @spec map_response({:done, reference()} | {atom(), reference(), any}) :: {:done | {atom(), any}, reference()} def map_response({:done, ref}), do: {:done, ref} def map_response({type, ref, value}), do: {{type, value}, ref} @spec flush_buffer(request()) :: request() def flush_buffer(%{waiting: waiting, response: response} = request) when not is_nil(waiting) and response != %{} do buffer = response |> Map.update(:data, nil, &(&1 |> Enum.reverse() |> IO.iodata_to_binary())) |> Map.update(:stdout, nil, &(&1 |> Enum.reverse() |> IO.iodata_to_binary())) |> Map.update(:stderr, nil, &(&1 |> Enum.reverse() |> IO.iodata_to_binary())) |> Map.update(:error, nil, &(&1 |> Enum.reverse() |> IO.iodata_to_binary())) |> Enum.to_list() |> Enum.reject(&(&1 |> elem(1) |> is_nil())) type = if is_map_key(response, :done) or is_map_key(response, :close), do: :halt, else: :cont GenServer.reply(request.waiting, {type, buffer}) struct!(request, response: %{}, waiting: nil) end def flush_buffer(request), do: request end