defmodule Kadabra.Connection.FlowControl do @moduledoc false @default_window_size round(:math.pow(2, 16) - 1) @max_window_size round(:math.pow(2, 31) - 1) @default_initial_window_size round(:math.pow(2, 16) - 1) @default_max_frame_size round(:math.pow(2, 14)) defstruct queue: :queue.new(), stream_set: %Kadabra.StreamSet{}, initial_window_size: @default_initial_window_size, max_frame_size: @default_max_frame_size, window: @default_window_size alias Kadabra.{Config, StreamSet, StreamSupervisor} @type t :: %__MODULE__{ queue: :queue.queue(), stream_set: StreamSet.t(), initial_window_size: non_neg_integer, max_frame_size: non_neg_integer, window: integer } def window_default, do: @default_window_size def window_max, do: @max_window_size @spec update_settings(t(), integer, integer, integer) :: t() def update_settings(flow_control, initial_window, max_frame, max_streams) do new_set = Map.put(flow_control.stream_set, :max_concurrent_streams, max_streams) flow_control |> Map.put(:initial_window_size, initial_window) |> Map.put(:max_frame_size, max_frame) |> Map.put(:stream_set, new_set) 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""" Adds new sendable item to the queue. ## Examples iex> flow = add(%Kadabra.Connection.FlowControl{}, %Kadabra.Request{}) iex> :queue.len(flow.queue) 1 """ @spec add(t, Kadabra.Request.t()) :: t def add(%{queue: queue} = flow_control, requests) when is_list(requests) do queue = Enum.reduce(requests, queue, &:queue.in(&1, &2)) %{flow_control | queue: queue} end def add(%{queue: queue} = flow_control, request) do queue = :queue.in(request, queue) %{flow_control | queue: queue} end def add_active(%{stream_set: set} = flow_control, stream_id) do new_set = StreamSet.add_active(set, stream_id) %{flow_control | stream_set: new_set} end @spec process(t, Config.t()) :: t def process(%{queue: queue, stream_set: stream_set} = flow, config) do with {{:value, request}, queue} <- :queue.out(queue), {:can_send, true} <- {:can_send, StreamSet.can_send?(stream_set)}, {:can_send, true} <- {:can_send, can_send?(flow)} do %{ stream_set: %{stream_id: stream_id}, initial_window_size: window, max_frame_size: max_frame } = flow case StreamSupervisor.start_stream(config, stream_id, window, max_frame) do {:ok, pid} -> Process.monitor(pid) size = byte_size(request.body || <<>>) :gen_statem.call(pid, {:send_headers, request}) updated_set = add_stream(stream_set, stream_id) flow |> Map.put(:queue, queue) |> Map.put(:stream_set, updated_set) |> decrement_window(size) |> process(config) other -> raise "something happened #{inspect(other)}" flow end else {:empty, _queue} -> flow {:can_send, false} -> flow end end defp add_stream(stream_set, stream_id) do stream_set |> StreamSet.add_active(stream_id) |> StreamSet.increment_active_stream_count() |> StreamSet.increment_stream_id() end @spec finish_stream(t, non_neg_integer) :: t def finish_stream(%{stream_set: set} = flow_control, stream_id) do new_set = set |> StreamSet.decrement_active_stream_count() |> StreamSet.remove_active(stream_id) %{flow_control | stream_set: new_set} end @doc ~S""" Returns true if window is positive. ## Examples iex> flow = %Kadabra.Connection.FlowControl{window: 500} iex> can_send?(flow) true iex> flow = %Kadabra.Connection.FlowControl{window: 0} iex> can_send?(flow) false """ @spec can_send?(t) :: boolean def can_send?(%{window: bytes}) when bytes > 0, do: true def can_send?(_else), do: false end