defmodule Kadabra.Stream.FlowControl do @moduledoc false alias Kadabra.Packetizer @default_max_frame round(:math.pow(2, 14)) @default_window round(:math.pow(2, 16) - 1) defstruct queue: :queue.new(), out_queue: :queue.new(), window: @default_window, max_frame_size: @default_max_frame @type t :: %__MODULE__{ max_frame_size: non_neg_integer, queue: :queue.queue(binary), out_queue: :queue.queue({binary, boolean}), window: integer } @type sock :: {:sslsocket, any, pid | {any, any}} @type frame :: {:send, binary} @doc ~S""" Returns new `Kadabra.Stream.FlowControl` with given opts. ## Examples iex> new() %Kadabra.Stream.FlowControl{} iex> new(window: 20_000, max_frame_size: 18_000) %Kadabra.Stream.FlowControl{window: 20_000, max_frame_size: 18_000} """ @spec new(Keyword.t()) :: t def new(opts \\ []) do %__MODULE__{ window: Keyword.get(opts, :window, @default_window), max_frame_size: Keyword.get(opts, :max_frame_size, @default_max_frame) } end @doc ~S""" Enqueues a sendable payload. ## Examples iex> add(%Kadabra.Stream.FlowControl{}, "test") %Kadabra.Stream.FlowControl{queue: {["test"], []}} """ @spec add(t, binary) :: t def add(flow_control, bin) do queue = :queue.in(bin, flow_control.queue) Map.put(flow_control, :queue, queue) end @doc ~S""" Processes sendable data in queue, if any present and window is positive. ## Examples iex> process(%Kadabra.Stream.FlowControl{queue: :queue.new()}) %Kadabra.Stream.FlowControl{queue: {[], []}} iex> queue = :queue.in({:send, "test"}, :queue.new()) iex> process(%Kadabra.Stream.FlowControl{queue: queue, window: -20}) %Kadabra.Stream.FlowControl{queue: {[send: "test"], []}, window: -20} """ @spec process(t) :: t def process(%{window: window} = flow_control) when window <= 0 do flow_control end def process(%{queue: queue} = flow_control) do case :queue.out(queue) do {{:value, bin}, queue} -> flow_control |> Map.put(:queue, queue) |> do_process(bin) {:empty, _queue} -> flow_control end end defp do_process(%{window: window} = flow, bin) when byte_size(bin) > window do %{ queue: queue, out_queue: out_queue, max_frame_size: max_size } = flow {chunk, rem_bin} = :erlang.split_binary(bin, window) payloads = Packetizer.split(max_size, chunk) out_queue = enqueue_partial(out_queue, payloads) queue = :queue.in_r(rem_bin, queue) flow |> Map.put(:queue, queue) |> Map.put(:out_queue, out_queue) |> Map.put(:window, 0) |> process() end defp do_process(%{window: window} = flow_control, bin) do %{max_frame_size: max_size, out_queue: out_queue} = flow_control payloads = Packetizer.split(max_size, bin) out_queue = enqueue_complete(out_queue, payloads) flow_control |> Map.put(:window, window - byte_size(bin)) |> Map.put(:out_queue, out_queue) |> process() end @doc ~S""" Increments stream window by given increment. ## Examples iex> increment_window(%Kadabra.Stream.FlowControl{window: 0}, 736) %Kadabra.Stream.FlowControl{window: 736} """ @spec increment_window(t, pos_integer) :: t def increment_window(flow_control, amount) do %{flow_control | window: flow_control.window + amount} end @doc ~S""" Sets stream max_frame_size to given size. ## Examples iex> set_max_frame_size(%Kadabra.Stream.FlowControl{ ...> max_frame_size: 16_384}, 1_040_200) %Kadabra.Stream.FlowControl{max_frame_size: 1_040_200} """ @spec set_max_frame_size(t, pos_integer) :: t def set_max_frame_size(flow_control, size) do %{flow_control | max_frame_size: size} end defp enqueue_complete(queue, []), do: queue defp enqueue_complete(queue, [payload | []]) do {payload, true} |> :queue.in(queue) |> enqueue_complete([]) end defp enqueue_complete(queue, [payload | rest]) do {payload, false} |> :queue.in(queue) |> enqueue_complete(rest) end defp enqueue_partial(queue, []), do: queue defp enqueue_partial(queue, [payload | rest]) do {payload, false} |> :queue.in(queue) |> enqueue_partial(rest) end end