defmodule Kadabra.Stream.FlowControl do @moduledoc false defstruct queue: :queue.new(), window: 56_536, max_frame_size: 16_384, stream_id: nil alias Kadabra.{Encodable, Frame} alias Kadabra.Connection.Socket @type t :: %__MODULE__{ max_frame_size: non_neg_integer, queue: :queue.queue(binary), 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(stream_id: 1) %Kadabra.Stream.FlowControl{stream_id: 1} iex> new(stream_id: 1, window: 20_000, max_frame_size: 18_000) %Kadabra.Stream.FlowControl{stream_id: 1, window: 20_000, max_frame_size: 18_000} """ @spec new(Keyword.t()) :: t def new(opts \\ []) do %__MODULE__{ stream_id: opts[:stream_id], window: Keyword.get(opts, :window, 56_536), max_frame_size: Keyword.get(opts, :max_frame_size, 16_384) } 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()}, self()) %Kadabra.Stream.FlowControl{queue: {[], []}} iex> queue = :queue.in({:send, "test"}, :queue.new()) iex> process(%Kadabra.Stream.FlowControl{queue: queue, ...> window: -20}, self()) %Kadabra.Stream.FlowControl{queue: {[send: "test"], []}, window: -20} """ @spec process(t, sock) :: t def process(%{window: window} = flow_control, _sock) when window <= 0 do flow_control end def process(%{queue: queue} = flow_control, socket) do case :queue.out(queue) do {{:value, bin}, queue} -> flow_control |> Map.put(:queue, queue) |> do_process(socket, bin) {:empty, _queue} -> flow_control end end def do_process(flow_control, socket, bin) do %{ queue: queue, max_frame_size: max_size, window: window, stream_id: stream_id } = flow_control size = byte_size(bin) if size > window do {chunk, rem_bin} = :erlang.split_binary(bin, window) max_size |> split_packet(chunk) |> send_partial_data(socket, stream_id) queue = :queue.in_r(rem_bin, queue) flow_control |> Map.put(:queue, queue) |> Map.put(:window, 0) |> process(socket) else max_size |> split_packet(bin) |> send_data(socket, stream_id) flow_control |> Map.put(:window, window - size) |> process(socket) end end def send_partial_data([], _socket, _stream_id), do: :ok def send_partial_data([bin | rest], socket, stream_id) do p = %Frame.Data{stream_id: stream_id, end_stream: false, data: bin} |> Encodable.to_bin() Socket.send(socket, p) send_partial_data(rest, socket, stream_id) end def send_data([], _socket, _stream_id), do: :ok def send_data([bin | []], socket, stream_id) do p = %Frame.Data{stream_id: stream_id, end_stream: true, data: bin} |> Encodable.to_bin() Socket.send(socket, p) send_data([], socket, stream_id) end def send_data([bin | rest], socket, stream_id) do p = %Frame.Data{stream_id: stream_id, end_stream: false, data: bin} |> Encodable.to_bin() Socket.send(socket, p) send_data(rest, socket, stream_id) end def split_packet(size, p) when byte_size(p) >= size do {chunk, rest} = :erlang.split_binary(p, size) [chunk | split_packet(size, rest)] end def split_packet(_size, <<>>), do: [] def split_packet(_size, p), do: [p] @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 end