defmodule Kadabra.Connection.FlowControl do @moduledoc false defstruct queue: [], stream_id: 1, active_stream_count: 0, active_streams: MapSet.new, window: 65_535, settings: %Kadabra.Connection.Settings{} alias Kadabra.Connection @type t :: %__MODULE__{ queue: [...], stream_id: pos_integer, active_stream_count: non_neg_integer, active_streams: MapSet.t, window: integer, settings: Connection.Settings.t } @spec update_settings(t, Connection.Settings.t) :: t def update_settings(%{settings: old_settings} = flow_control, settings) do settings = Connection.Settings.merge(old_settings, settings) %{flow_control | settings: settings} end @doc ~S""" Increments current `stream_id`. ## Examples iex> flow = %Kadabra.Connection.FlowControl{stream_id: 5} iex> increment_stream_id(flow) %Kadabra.Connection.FlowControl{stream_id: 7} """ @spec increment_stream_id(t) :: t def increment_stream_id(flow_control) do %{flow_control | stream_id: flow_control.stream_id + 2} end @doc ~S""" Increments open stream count. ## Examples iex> flow = %Kadabra.Connection.FlowControl{active_stream_count: 2} iex> increment_active_stream_count(flow) %Kadabra.Connection.FlowControl{active_stream_count: 3} """ @spec increment_active_stream_count(t) :: t def increment_active_stream_count(flow_control) do %{flow_control | active_stream_count: flow_control.active_stream_count + 1} end @doc ~S""" Decrements open stream count. ## Examples iex> flow = %Kadabra.Connection.FlowControl{active_stream_count: 2} iex> decrement_active_stream_count(flow) %Kadabra.Connection.FlowControl{active_stream_count: 1} """ @spec decrement_active_stream_count(t) :: t def decrement_active_stream_count(flow_control) do %{flow_control | active_stream_count: flow_control.active_stream_count - 1} end @doc ~S""" Increments available window. ## Examples iex> flow = %Kadabra.Connection.FlowControl{window: 1_000} iex> increment_window(flow, 500) %Kadabra.Connection.FlowControl{window: 1_500} """ @spec increment_window(t, pos_integer) :: t def increment_window(%{window: window} = flow_control, amount) do %{flow_control | window: window + amount} end @doc ~S""" Decrements available window. ## Examples iex> flow = %Kadabra.Connection.FlowControl{window: 1_000} iex> decrement_window(flow, 500) %Kadabra.Connection.FlowControl{window: 500} """ @spec decrement_window(t, pos_integer) :: t def decrement_window(%{window: window} = flow_control, amount) do %{flow_control | window: window - amount} end @doc ~S""" Marks stream_id as active. ## Examples iex> flow = add_active(%Kadabra.Connection.FlowControl{}, 1) iex> flow.active_streams #MapSet<[1]> """ def add_active(%{active_streams: active} = flow_control, stream_id) do %{flow_control | active_streams: MapSet.put(active, stream_id)} end @doc ~S""" Marks stream_id as active. ## Examples iex> flow = remove_active(%Kadabra.Connection.FlowControl{ ...> active_streams: MapSet.new([1, 3])}, 1) iex> flow.active_streams #MapSet<[3]> """ def remove_active(%{active_streams: active} = flow_control, stream_id) do %{flow_control | active_streams: MapSet.delete(active, stream_id)} end @doc ~S""" Adds new sendable item to the queue. ## Examples iex> flow = %Kadabra.Connection.FlowControl{queue: []} iex> add(flow, "test", "payload") %Kadabra.Connection.FlowControl{queue: [{:send, "test", "payload"}]} """ @spec add(t, [...], binary | nil) :: t def add(%{queue: queue} = flow_control, headers, payload \\ nil) do %{flow_control | queue: queue ++ [{:send, headers, payload}]} end @spec process(t, Connection.t) :: t def process(%{queue: []} = flow_control, _connection) do flow_control end def process(%{queue: [{:send, headers, payload} | rest]} = flow, conn) do if can_send?(flow) do {:ok, pid} = Kadabra.Supervisor.start_stream(conn) size = byte_size(payload || <<>>) :gen_statem.call(pid, {:send_headers, headers, payload}) flow_control = %{flow | queue: rest} flow_control |> decrement_window(size) |> add_active(flow.stream_id) |> increment_active_stream_count() |> increment_stream_id() else flow end end @doc ~S""" Returns true if active_streams is less than max streams and window is positive. ## Examples iex> settings = %Kadabra.Connection.Settings{max_concurrent_streams: 100} iex> flow = %Kadabra.Connection.FlowControl{active_stream_count: 3, ...> window: 500, settings: settings} iex> can_send?(flow) true iex> settings = %Kadabra.Connection.Settings{max_concurrent_streams: 100} iex> flow = %Kadabra.Connection.FlowControl{active_stream_count: 3, ...> window: 0, settings: settings} iex> can_send?(flow) false iex> settings = %Kadabra.Connection.Settings{max_concurrent_streams: 1} iex> flow = %Kadabra.Connection.FlowControl{active_stream_count: 3, ...> window: 500, settings: settings} iex> can_send?(flow) false """ @spec can_send?(t) :: boolean def can_send?(%{active_stream_count: count, settings: settings, window: bytes}) do count < settings.max_concurrent_streams and bytes > 0 end end