# frozen_string_literal: true

module HTTP2
  # Default connection and stream flow control window (64KB).
  DEFAULT_FLOW_WINDOW = 65_535

  # Default header table size
  DEFAULT_HEADER_SIZE = 4096

  # Default stream_limit
  DEFAULT_MAX_CONCURRENT_STREAMS = 100

  # Default values for SETTINGS frame, as defined by the spec.
  SPEC_DEFAULT_CONNECTION_SETTINGS = {
    settings_header_table_size: 4096,
    settings_enable_push: 1, # enabled for servers
    settings_max_concurrent_streams: Framer::MAX_STREAM_ID, # unlimited
    settings_initial_window_size: 65_535,
    settings_max_frame_size: 16_384,
    settings_max_header_list_size: (2 << 30) - 1 # unlimited
  }.freeze

  DEFAULT_CONNECTION_SETTINGS = {
    settings_header_table_size: 4096,
    settings_enable_push: 1, # enabled for servers
    settings_max_concurrent_streams: 100,
    settings_initial_window_size: 65_535,
    settings_max_frame_size: 16_384,
    settings_max_header_list_size: (2 << 30) - 1 # unlimited
  }.freeze

  # Default stream priority (lower values are higher priority).
  DEFAULT_WEIGHT = 16

  # Default connection "fast-fail" preamble string as defined by the spec.
  CONNECTION_PREFACE_MAGIC = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"

  REQUEST_MANDATORY_HEADERS = %w[:scheme :method :authority :path].freeze

  RESPONSE_MANDATORY_HEADERS = %w[:status].freeze

  CONNECTION_FRAME_TYPES = %i[settings ping goaway].freeze

  HEADERS_FRAME_TYPES = %i[headers push_promise].freeze

  STREAM_OPEN_STATES = %i[open half_closed_local].freeze

  # Connection encapsulates all of the connection, stream, flow-control,
  # error management, and other processing logic required for a well-behaved
  # HTTP 2.0 endpoint.
  #
  # Note that this class should not be used directly. Instead, you want to
  # use either Client or Server class to drive the HTTP 2.0 exchange.
  #
  class Connection
    include FlowBuffer
    include Emitter
    include Error
    include BufferUtils

    # Connection state (:new, :closed).
    attr_reader :state

    # Size of current connection flow control window (by default, set to
    # infinity, but is automatically updated on receipt of peer settings).
    attr_reader :local_window
    attr_reader :remote_window, :remote_settings
    alias window local_window

    # Current settings value for local and peer
    attr_reader :local_settings

    # Pending settings value
    #  Sent but not ack'ed settings
    attr_reader :pending_settings

    # Number of active streams between client and server (reserved streams
    # are not counted towards the stream limit).
    attr_accessor :active_stream_count

    # Initializes new connection object.
    #
    def initialize(settings = {})
      @local_settings  = DEFAULT_CONNECTION_SETTINGS.merge(settings)
      @remote_settings = SPEC_DEFAULT_CONNECTION_SETTINGS.dup

      @compressor   = Header::Compressor.new(settings)
      @decompressor = Header::Decompressor.new(settings)

      @active_stream_count = 0
      @last_stream_id = 0
      @streams = {}
      @streams_recently_closed = {}
      @pending_settings = []

      @framer = Framer.new(@local_settings[:settings_max_frame_size])

      @local_window_limit = @local_settings[:settings_initial_window_size]
      @local_window = @local_window_limit
      @remote_window_limit = @remote_settings[:settings_initial_window_size]
      @remote_window = @remote_window_limit

      @recv_buffer = "".b
      @continuation = []
      @error = nil

      @h2c_upgrade = nil
      @closed_since = nil
      @received_frame = false

      # from mixins
      @listeners = Hash.new { |hash, key| hash[key] = [] }
      @send_buffer = FrameBuffer.new
    end

    def closed?
      @state == :closed
    end

    # Allocates new stream for current connection.
    #
    # @param priority [Integer]
    # @param window [Integer]
    # @param parent [Stream]
    def new_stream(**args)
      raise ConnectionClosed if @state == :closed
      raise StreamLimitExceeded if @active_stream_count >= @remote_settings[:settings_max_concurrent_streams]

      connection_error(:protocol_error, msg: "id is smaller than previous") if @stream_id < @last_stream_id

      stream = activate_stream(id: @stream_id, **args)
      @last_stream_id = stream.id

      @stream_id += 2

      stream
    end

    # Sends PING frame to the peer.
    #
    # @param payload [String] optional payload must be 8 bytes long
    # @param blk [Proc] callback to execute when PONG is received
    def ping(payload, &blk)
      send(type: :ping, stream: 0, payload: payload)
      once(:ack, &blk) if blk
    end

    # Sends a GOAWAY frame indicating that the peer should stop creating
    # new streams for current connection.
    #
    # Endpoints MAY append opaque data to the payload of any GOAWAY frame.
    # Additional debug data is intended for diagnostic purposes only and
    # carries no semantic value. Debug data MUST NOT be persistently stored,
    # since it could contain sensitive information.
    #
    # @param error [Symbol]
    # @param payload [String]
    def goaway(error = :no_error, payload = nil)
      send(type: :goaway, last_stream: @last_stream_id,
           error: error, payload: payload)
      @state = :closed
      @closed_since = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    end

    # Sends a WINDOW_UPDATE frame to the peer.
    #
    # @param increment [Integer]
    def window_update(increment)
      @local_window += increment
      send(type: :window_update, stream: 0, increment: increment)
    end

    # Sends a connection SETTINGS frame to the peer.
    # The values are reflected when the corresponding ACK is received.
    #
    # @param settings [Array or Hash]
    def settings(payload)
      validate_settings(@local_role, payload)
      @pending_settings << payload
      send(type: :settings, stream: 0, payload: payload)
      @pending_settings << payload
    end

    # Decodes incoming bytes into HTTP 2.0 frames and routes them to
    # appropriate receivers: connection frames are handled directly, and
    # stream frames are passed to appropriate stream objects.
    #
    # @param data [String] Binary encoded string
    def receive(data)
      append_str(@recv_buffer, data)

      # Upon establishment of a TCP connection and determination that
      # HTTP/2.0 will be used by both peers, each endpoint MUST send a
      # connection header as a final confirmation and to establish the
      # initial settings for the HTTP/2.0 connection.
      #
      # Client connection header is 24 byte connection header followed by
      # SETTINGS frame. Server connection header is SETTINGS frame only.
      if @state == :waiting_magic
        if @recv_buffer.size < 24
          raise HandshakeError unless CONNECTION_PREFACE_MAGIC.start_with? @recv_buffer

          return # maybe next time
        elsif read_str(@recv_buffer, 24) == CONNECTION_PREFACE_MAGIC
          # MAGIC is OK.  Send our settings
          @state = :waiting_connection_preface
          payload = @local_settings.reject { |k, v| v == SPEC_DEFAULT_CONNECTION_SETTINGS[k] }
          settings(payload)
        else
          raise HandshakeError
        end
      end

      while (frame = @framer.parse(@recv_buffer))
        # @type var stream_id: Integer
        stream_id = frame[:stream]
        frame_type = frame[:type]

        if is_a?(Client) && !@received_frame
          connection_error(:protocol_error, msg: "didn't receive settings") if frame_type != :settings
          @received_frame = true
        end

        # Implementations MUST discard frames
        # that have unknown or unsupported types.
        if frame_type.nil?
          # However, extension frames that appear in
          # the middle of a header block (Section 4.3) are not permitted; these
          # MUST be treated as a connection error (Section 5.4.1) of type
          # PROTOCOL_ERROR.
          connection_error(:protocol_error) unless @continuation.empty?
          next
        end

        emit(:frame_received, frame)

        # Header blocks MUST be transmitted as a contiguous sequence of frames
        # with no interleaved frames of any other type, or from any other stream.
        unless @continuation.empty?
          connection_error unless frame_type == :continuation && stream_id == @continuation.first[:stream]

          @continuation << frame
          unless frame[:flags].include? :end_headers
            buffered_payload = @continuation.sum { |f| f[:payload].bytesize }
            # prevent HTTP/2 CONTINUATION FLOOD
            # same heuristic as the one from HAProxy: https://www.haproxy.com/blog/haproxy-is-resilient-to-the-http-2-continuation-flood
            # different mitigation (connection closed, instead of 400 response)
            unless buffered_payload < @local_settings[:settings_max_frame_size]
              connection_error(:protocol_error,
                               msg: "too many continuations received")
            end

            next
          end

          payload = @continuation.map { |f| f[:payload] }.join

          frame = @continuation.shift
          frame_type = frame[:type]

          @continuation.clear

          frame.delete(:length)
          frame[:payload] = payload
          frame[:flags] << :end_headers
        end

        # SETTINGS frames always apply to a connection, never a single stream.
        # The stream identifier for a settings frame MUST be zero.  If an
        # endpoint receives a SETTINGS frame whose stream identifier field is
        # anything other than 0x0, the endpoint MUST respond with a connection
        # error (Section 5.4.1) of type PROTOCOL_ERROR.
        if connection_frame?(frame)
          connection_error(:protocol_error) unless stream_id.zero?
          connection_management(frame)
        else
          case frame_type
          when :headers
            # When server receives even-numbered stream identifier,
            # the endpoint MUST respond with a connection error of type PROTOCOL_ERROR.
            connection_error if stream_id.even? && is_a?(Server)

            # The last frame in a sequence of HEADERS/CONTINUATION
            # frames MUST have the END_HEADERS flag set.
            unless frame[:flags].include? :end_headers
              @continuation << frame
              next
            end

            # After sending a GOAWAY frame, the sender can discard frames
            # for new streams.  However, any frames that alter connection
            # state cannot be completely ignored.  For instance, HEADERS,
            # PUSH_PROMISE and CONTINUATION frames MUST be minimally
            # processed to ensure a consistent compression state
            decode_headers(frame)
            return if @state == :closed

            stream = @streams[stream_id]
            if stream.nil?
              verify_pseudo_headers(frame)

              verify_stream_order(stream_id)
              stream = activate_stream(
                id: stream_id,
                weight: frame[:weight] || DEFAULT_WEIGHT,
                dependency: frame[:dependency] || 0,
                exclusive: frame[:exclusive] || false
              )
              emit(:stream, stream)
            end

            stream << frame

          when :push_promise
            # The last frame in a sequence of PUSH_PROMISE/CONTINUATION
            # frames MUST have the END_HEADERS flag set
            unless frame[:flags].include? :end_headers
              @continuation << frame
              return
            end

            decode_headers(frame)
            return if @state == :closed

            # PUSH_PROMISE frames MUST be associated with an existing, peer-
            # initiated stream... A receiver MUST treat the receipt of a
            # PUSH_PROMISE on a stream that is neither "open" nor
            # "half-closed (local)" as a connection error (Section 5.4.1) of
            # type PROTOCOL_ERROR. Similarly, a receiver MUST treat the
            # receipt of a PUSH_PROMISE that promises an illegal stream
            # identifier (Section 5.1.1) (that is, an identifier for a stream
            # that is not currently in the "idle" state) as a connection error
            # (Section 5.4.1) of type PROTOCOL_ERROR, unless the receiver
            # recently sent a RST_STREAM frame to cancel the associated stream.
            parent = @streams[stream_id]
            pid = frame[:promise_stream]

            # if PUSH parent is recently closed, RST_STREAM the push
            if @streams_recently_closed[stream_id]
              send(type: :rst_stream, stream: pid, error: :refused_stream)
              return
            end

            connection_error(msg: "missing parent ID") if parent.nil?

            unless STREAM_OPEN_STATES.include?(parent.state)
              # An endpoint might receive a PUSH_PROMISE frame after it sends
              # RST_STREAM.  PUSH_PROMISE causes a stream to become "reserved".
              # The RST_STREAM does not cancel any promised stream.  Therefore, if
              # promised streams are not desired, a RST_STREAM can be used to
              # close any of those streams.
              if parent.closed == :local_rst
                # We can either (a) 'resurrect' the parent, or (b) RST_STREAM
                # ... sticking with (b), might need to revisit later.
                send(type: :rst_stream, stream: pid, error: :refused_stream)
              else
                connection_error
              end
            end

            _verify_pseudo_headers(frame, REQUEST_MANDATORY_HEADERS)
            verify_stream_order(pid)
            stream = activate_stream(id: pid, parent: parent)
            emit(:promise, stream)
            stream << frame
          else
            if (stream = @streams[stream_id])
              stream << frame
              if frame_type == :data
                update_local_window(frame)
                calculate_window_update(@local_window_limit)
              end
            else
              case frame_type
              # The PRIORITY frame can be sent for a stream in the "idle" or
              # "closed" state. This allows for the reprioritization of a
              # group of dependent streams by altering the priority of an
              # unused or closed parent stream.
              when :priority
                stream = activate_stream(
                  id: stream_id,
                  weight: frame[:weight] || DEFAULT_WEIGHT,
                  dependency: frame[:dependency] || 0,
                  exclusive: frame[:exclusive] || false
                )

                emit(:stream, stream)
                stream << frame

              # WINDOW_UPDATE can be sent by a peer that has sent a frame
              # bearing the END_STREAM flag. This means that a receiver could
              # receive a WINDOW_UPDATE frame on a "half-closed (remote)" or
              # "closed" stream. A receiver MUST NOT treat this as an error
              # (see Section 5.1).
              when :window_update
                unless @streams_recently_closed.key?(stream_id)
                  connection_error(:protocol_error, msg: "sent window update on idle stream")
                end
                stream = @streams_recently_closed[stream_id]
                process_window_update(frame: frame, encode: true)
              # Endpoints MUST ignore
              # WINDOW_UPDATE or RST_STREAM frames received in this state (closed), though
              # endpoints MAY choose to treat frames that arrive a significant
              # time after sending END_STREAM as a connection error.
              when :rst_stream
                unless @streams_recently_closed.key?(stream_id)
                  connection_error(:protocol_error, msg: "sent window update on idle stream")
                end
              else
                # An endpoint that receives an unexpected stream identifier
                # MUST respond with a connection error of type PROTOCOL_ERROR.
                connection_error(msg: "stream does not exist")
              end
            end
          end
        end
      end
    rescue StandardError => e
      raise if e.is_a?(Error::Error)

      connection_error(e: e)
    end

    def <<(data)
      receive(data)
    end

    private

    # Send an outgoing frame. DATA frames are subject to connection flow
    # control and may be split and / or buffered based on current window size.
    # All other frames are sent immediately.
    #
    # @note all frames are currently delivered in FIFO order.
    # @param frame [Hash]
    def send(frame)
      frame_type = frame[:type]

      emit(:frame_sent, frame)
      if frame_type == :data
        send_data(frame, true)

      elsif frame_type == :rst_stream && frame[:error] == :protocol_error
        # An endpoint can end a connection at any time. In particular, an
        # endpoint MAY choose to treat a stream error as a connection error.

        goaway(:protocol_error)
      else
        # HEADERS and PUSH_PROMISE may generate CONTINUATION. Also send
        # RST_STREAM that are not protocol errors
        encode(frame)
      end
    end

    # Applies HTTP 2.0 binary encoding to the frame.
    #
    # @param frame [Hash]
    def encode(frame)
      if HEADERS_FRAME_TYPES.include?(frame[:type])
        encode_headers(frame) # HEADERS and PUSH_PROMISE may create more than one frame
      else
        emit(:frame, @framer.generate(frame))
      end
    end

    # Check if frame is a connection frame: SETTINGS, PING, GOAWAY, and any
    # frame addressed to stream ID = 0.
    #
    # @param frame [Hash]
    # @return [Boolean]
    def connection_frame?(frame)
      frame[:stream].zero? || CONNECTION_FRAME_TYPES.include?(frame[:type])
    end

    # Process received connection frame (stream ID = 0).
    # - Handle SETTINGS updates
    # - Connection flow control (WINDOW_UPDATE)
    # - Emit PONG auto-reply to PING frames
    # - Mark connection as closed on GOAWAY
    #
    # @param frame [Hash]
    def connection_management(frame)
      frame_type = frame[:type]

      case @state
      when :waiting_connection_preface
        # The first frame MUST be a SETTINGS frame at the start of a connection.
        connection_error unless frame[:type] == :settings

        @state = :connected
        connection_settings(frame)

      when :connected
        case frame_type
        when :settings
          connection_settings(frame)
        when :window_update
          process_window_update(frame: frame, encode: true)
        when :ping
          ping_management(frame)
        when :goaway
          # Receivers of a GOAWAY frame MUST NOT open additional streams on
          # the connection, although a new connection can be established
          # for new streams.
          @state = :closed
          @closed_since = Process.clock_gettime(Process::CLOCK_MONOTONIC)
          emit(:goaway, frame[:last_stream], frame[:error], frame[:payload])
        when :altsvc
          origin = frame[:origin]
          # 4.  The ALTSVC HTTP/2 Frame
          # An ALTSVC frame on stream 0 with empty (length 0) "Origin"
          # information is invalid and MUST be ignored.
          emit(:altsvc, frame) if origin && !origin.empty?
        when :origin
          return if @h2c_upgrade || !frame[:flags].empty?

          frame[:payload].each do |orig|
            emit(:origin, orig)
          end
        when :blocked
          emit(:blocked, frame)
        else
          connection_error
        end
      when :closed
        case frame_type
        when :goaway
          connection_error
        when :ping
          ping_management(frame)
        else
          connection_error if (Process.clock_gettime(Process::CLOCK_MONOTONIC) - @closed_since) > 15
        end
      else
        connection_error
      end
    end

    def ping_management(frame)
      if frame[:flags].include? :ack
        emit(:ack, frame[:payload])
      else
        send(type: :ping, stream: 0,
             flags: [:ack], payload: frame[:payload])
      end
    end

    # Validate settings parameters.  See sepc Section 6.5.2.
    #
    # @param role [Symbol] The sender's role: :client or :server
    # @return nil if no error.  Exception object in case of any error.
    def validate_settings(role, settings)
      settings.each do |key, v|
        case key
        when :settings_enable_push
          case role
          when :server
            # Section 8.2
            # Clients MUST reject any attempt to change the
            # SETTINGS_ENABLE_PUSH setting to a value other than 0 by treating the
            # message as a connection error (Section 5.4.1) of type PROTOCOL_ERROR.
            next if v.zero?

            connection_error(:protocol_error, msg: "invalid #{key} value")
          when :client
            # Any value other than 0 or 1 MUST be treated as a
            # connection error (Section 5.4.1) of type PROTOCOL_ERROR.
            next if v.zero? || v == 1

            connection_error(:protocol_error, msg: "invalid #{key} value")
          end
        when :settings_initial_window_size
          # Values above the maximum flow control window size of 2^31-1 MUST
          # be treated as a connection error (Section 5.4.1) of type
          # FLOW_CONTROL_ERROR.
          next if v <= 0x7fffffff

          connection_error(:flow_control_error, msg: "invalid #{key} value")
        when :settings_max_frame_size
          # The initial value is 2^14 (16,384) octets.  The value advertised
          # by an endpoint MUST be between this initial value and the maximum
          # allowed frame size (2^24-1 or 16,777,215 octets), inclusive.
          # Values outside this range MUST be treated as a connection error
          # (Section 5.4.1) of type PROTOCOL_ERROR.
          next if v.between?(16_384, 16_777_215)

          connection_error(:protocol_error, msg: "invalid #{key} value")
          # when :settings_max_concurrent_streams
          # Any value is valid
          # when :settings_header_table_size
          # Any value is valid
          # when :settings_max_header_list_size
          # Any value is valid
          # else # ignore unknown settings
        end
      end
    end

    # Update connection settings based on parameters set by the peer.
    #
    # @param frame [Hash]
    def connection_settings(frame)
      # Apply settings.
      #  side =
      #   local: previously sent and pended our settings should be effective
      #   remote: just received peer settings should immediately be effective
      if frame[:flags].include?(:ack)
        # Process pending settings we have sent.
        settings = @pending_settings.shift
        side = :local
      else
        validate_settings(@remote_role, frame[:payload])
        settings = frame[:payload]
        side = :remote
      end

      settings.each do |key, v|
        case side
        when :local
          @local_settings[key] = v
        when :remote
          @remote_settings[key] = v
        end

        case key
        when :settings_max_concurrent_streams
          # Do nothing.
          # The value controls at the next attempt of stream creation.

        when :settings_initial_window_size
          # A change to SETTINGS_INITIAL_WINDOW_SIZE could cause the available
          # space in a flow control window to become negative. A sender MUST
          # track the negative flow control window, and MUST NOT send new flow
          # controlled frames until it receives WINDOW_UPDATE frames that cause
          # the flow control window to become positive.
          case side
          when :local
            @local_window = @local_window - @local_window_limit + v
            @streams.each_value do |stream|
              stream.emit(:local_window, stream.local_window - @local_window_limit + v)
            end

            @local_window_limit = v
          when :remote
            # can adjust the initial window size for new streams by including a
            # value for SETTINGS_INITIAL_WINDOW_SIZE in the SETTINGS frame.
            # The connection flow-control window can only be changed using
            # WINDOW_UPDATE frames.
            @streams.each_value do |stream|
              # Event name is :window, not :remote_window
              stream.emit(:window, stream.remote_window - @remote_window_limit + v)
            end

            @remote_window_limit = v
          end

        when :settings_header_table_size
          # Setting header table size might cause some headers evicted
          case side
          when :local
            @compressor.table_size = v
          when :remote
            @decompressor.table_size = v
          end

        when :settings_enable_push
          # nothing to do

        when :settings_max_frame_size
          @framer.remote_max_frame_size = v

          # else # ignore unknown settings
        end
      end

      case side
      when :local
        # Received a settings_ack.  Notify application layer.
        emit(:settings_ack, frame, @pending_settings.size)
      when :remote
        unless @state == :closed || @h2c_upgrade == :start
          # Send ack to peer
          send(type: :settings, stream: 0, payload: [], flags: [:ack])
          # when initial window size changes, we try to flush any buffered
          # data.
          @streams.each_value(&:flush)
        end
      end
    end

    # Decode headers payload and update connection decompressor state.
    #
    # The receiver endpoint reassembles the header block by concatenating
    # the individual fragments, then decompresses the block to reconstruct
    # the header set - aka, header payloads are buffered until END_HEADERS,
    # or an END_PROMISE flag is seen.
    #
    # @param frame [Hash]
    def decode_headers(frame)
      frame[:payload] = @decompressor.decode(frame[:payload], frame) if frame[:payload].is_a?(String)
    rescue CompressionError => e
      connection_error(:compression_error, e: e)
    rescue ProtocolError => e
      connection_error(:protocol_error, e: e)
    rescue StandardError => e
      connection_error(:internal_error, e: e)
    end

    # Encode headers payload and update connection compressor state.
    #
    # @param headers_frame [Hash]
    def encode_headers(headers_frame)
      payload = headers_frame[:payload]
      begin
        payload = headers_frame[:payload] = @compressor.encode(payload) unless payload.is_a?(String)
      rescue StandardError => e
        connection_error(:compression_error, e: e)
      end

      max_frame_size = @remote_settings[:settings_max_frame_size]

      # if single frame, return immediately
      if payload.bytesize <= max_frame_size
        emit(:frame, @framer.generate(headers_frame))
        return
      end

      # split into multiple CONTINUATION frames
      headers_frame[:flags].delete(:end_headers)
      headers_frame[:payload] = payload.byteslice(0, max_frame_size)
      payload = payload.byteslice(max_frame_size..-1)

      # emit first HEADERS frame
      emit(:frame, @framer.generate(headers_frame))

      loop do
        continuation_frame = headers_frame.merge(
          type: :continuation,
          flags: EMPTY,
          payload: payload.byteslice(0, max_frame_size)
        )

        payload = payload.byteslice(max_frame_size..-1)

        if payload.nil? || payload.empty?
          continuation_frame[:flags] = [:end_headers]
          emit(:frame, @framer.generate(continuation_frame))
          break
        end

        emit(:frame, @framer.generate(continuation_frame))
      end
    end

    # Activates new incoming or outgoing stream and registers appropriate
    # connection managemet callbacks.
    #
    # @param id [Integer]
    # @param priority [Integer]
    # @param window [Integer]
    # @param parent [Stream]
    def activate_stream(id:, **args)
      connection_error(msg: "Stream ID already exists") if @streams.key?(id)

      raise StreamLimitExceeded if @active_stream_count >= @local_settings[:settings_max_concurrent_streams]

      stream = Stream.new(connection: self, id: id, **args)

      stream.once(:close) do
        @streams.delete(id)

        # Store a reference to the closed stream, such that we can respond
        # to any in-flight frames while close is registered on both sides.
        # References to such streams will be purged whenever another stream
        # is closed, with a minimum of 15s RTT time window.
        now = Process.clock_gettime(Process::CLOCK_MONOTONIC)

        _, closed_since = @streams_recently_closed.first

        # forego recently closed recycling if empty or the first element
        # hasn't expired yet (it's ordered).
        if closed_since && (now - closed_since) > 15
          # discards all streams which have closed for a while.
          # TODO: use a drop_while! variant whenever there is one.
          @streams_recently_closed = @streams_recently_closed.drop_while do |_, since|
            (now - since) > 15
          end.to_h
        end

        @streams_recently_closed[id] = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      end

      stream.on(:frame, &method(:send))
      @streams[id] = stream
    end

    def verify_stream_order(id)
      return unless id.odd?

      connection_error(msg: "Stream ID smaller than previous") if @last_stream_id >= id
      @last_stream_id = id
    end

    def _verify_pseudo_headers(frame, mandatory_headers)
      headers = frame[:payload]
      return if headers.is_a?(String)

      pseudo_headers = headers.take_while do |field, value|
        # use this loop to validate pseudo-headers
        connection_error(:protocol_error, msg: "path is empty") if field == ":path" && value.empty?
        field.start_with?(":")
      end.map(&:first)
      return if mandatory_headers.size == pseudo_headers.size &&
                (mandatory_headers - pseudo_headers).empty?

      connection_error(:protocol_error, msg: "invalid pseudo-headers")
    end

    # Emit GOAWAY error indicating to peer that the connection is being
    # aborted, and once sent, raise a local exception.
    #
    # @param error [Symbol]
    # @option error [Symbol] :no_error
    # @option error [Symbol] :internal_error
    # @option error [Symbol] :flow_control_error
    # @option error [Symbol] :stream_closed
    # @option error [Symbol] :frame_too_large
    # @option error [Symbol] :compression_error
    # @param msg [String]
    def connection_error(error = :protocol_error, msg: nil, e: nil)
      goaway(error) unless @state == :closed || @state == :new

      @state = :closed
      @error = error
      msg ||= e ? e.message : "protocol error"
      backtrace = e ? e.backtrace : nil
      raise Error.types[error], msg, backtrace
    end
    alias error connection_error

    def manage_state(_)
      yield
    end
  end
end