defmodule Kadabra.Stream.FlowControl do @moduledoc false defstruct queue: [], window: 56_536, max_frame_size: 16_536, stream_id: nil alias Kadabra.{Encodable, Frame} @type t :: %__MODULE__{ max_frame_size: non_neg_integer, queue: [] | [...], 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} """ @spec new(Keyword.t) :: t def new(opts \\ []) do %__MODULE__{ stream_id: opts[:stream_id], window: opts[:window] || 56_536 } end @doc ~S""" Enqueues a sendable payload. ## Examples iex> add(%Kadabra.Stream.FlowControl{}, "test") %Kadabra.Stream.FlowControl{queue: [{:send, "test"}]} """ @spec add(t, binary) :: t def add(flow_control, bin) do queue = flow_control.queue ++ [{:send, bin}] %{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: []}, self()) %Kadabra.Stream.FlowControl{queue: []} iex> process(%Kadabra.Stream.FlowControl{queue: [{:send, "test"}], ...> window: -20}, self()) %Kadabra.Stream.FlowControl{queue: [{:send, "test"}], window: -20} """ @spec process(t, sock) :: t def process(%{queue: []} = flow_control, _sock) do flow_control end def process(%{window: window} = flow_control, _sock) when window <= 0 do flow_control end def process(%{queue: [{:send, bin} | rest], window: window, stream_id: stream_id} = flow_control, socket) do size = byte_size(bin) if size > window do {chunk, rem_bin} = :erlang.split_binary(bin, window) bin = %Frame.Data{stream_id: stream_id, end_stream: false, data: chunk} |> Encodable.to_bin :ssl.send(socket, bin) flow_control = %{flow_control | queue: [{:send, rem_bin} | rest], window: 0 } process(flow_control, socket) else bin = %Frame.Data{stream_id: stream_id, end_stream: true, data: bin} |> Encodable.to_bin :ssl.send(socket, bin) flow_control = %{flow_control | queue: rest, window: window - size} process(flow_control, socket) end 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 end