class ObjectFramework::HTTP2::Connection

# File lib/object_framework/http2/connection.rb, line 14
def initialize(io:, on_stream:, reporter:, header_limit: 32 * 1024,
               stream_limit: 128, idle_timeout: 30, write_timeout: 30,
               tls: false)
  @io, @on_stream, @reporter = io, on_stream, reporter
  @header_limit, @stream_limit, @idle_timeout, @write_timeout, @tls = header_limit, stream_limit, idle_timeout, write_timeout, tls
  @reader = HTTP::Reader.new(io:)
  @decoder, @encoder = HPACK::Decoder.new(header_limit:), HPACK::Encoder.new
  @mutex, @write_mutex, @changed = Mutex.new, Mutex.new, ConditionVariable.new
  @streams, @last_stream, @closed, @draining = {}, 0, false, false
  @peer_initial_window, @send_window, @receive_window = 65_535, 65_535, 65_535
  @peer_frame_size, @continuation, @first_frame = 16_384, nil, true
  @context = ExecutionContext.new
end

# File lib/object_framework/http2/connection.rb, line 59
def close
  streams = @mutex.synchronize do
    @closed = true
    @changed.broadcast
    @streams.values.dup
  end
  streams.each { |stream| stream.input.cancel }
  io.close unless io.closed?
rescue IOError, SystemCallError
  nil
end

# File lib/object_framework/http2/connection.rb, line 111
def reset?(stream)
  @mutex.synchronize { @closed || stream.reset }
end

# File lib/object_framework/http2/connection.rb, line 28
def run
  handshake if @tls
  raise ProtocolError, "Invalid HTTP/2 connection preface" unless @reader.read_exact(PREFACE.bytesize, deadline: deadline) == PREFACE
  send_frame(4, 0, 0, [3, @stream_limit, 6, @header_limit].pack("nNnN"))
  loop do
    frame = Frame.read(@reader, deadline:)
    if @first_frame
      raise ProtocolError unless frame.type == 4 && frame.stream_id.zero? && (frame.flags & 1).zero?
      @first_frame = false
    end
    raise ProtocolError if @continuation && (frame.type != 9 || frame.stream_id != @continuation[:id])
    begin
      process(frame)
    rescue StreamError => error
      reject_stream(error)
    end
  end
rescue EOFError, IOError, SystemCallError
  nil
rescue StandardError => error
  @reporter.report(error, context: @context)
  begin
    code = error.is_a?(ProtocolError) ? error.code : 2
    send_frame(7, 0, 0, [@last_stream, code].pack("NN"))
  rescue StandardError
    nil
  end
ensure
  close
end

# File lib/object_framework/http2/connection.rb, line 89
def send_data(stream, bytes)
  offset = 0
  while offset < bytes.bytesize
    count = @mutex.synchronize do
      loop do
        stream.context.cancellation.check!
        raise Cancelled, "HTTP/2 stream is closed" if @closed || stream.reset || stream.local_closed
        available = [@send_window, stream.send_window, @peer_frame_size, bytes.bytesize - offset].min
        if available.positive?
          @send_window -= available
          stream.send_window -= available
          break available
        end
        remaining = stream.context.cancellation.deadline ? stream.context.cancellation.deadline - HTTP::Timing.now : 0.1
        @changed.wait(@mutex, [[remaining, 0].max, 0.1].min)
      end
    end
    send_frame(0, 0, stream.id, bytes.byteslice(offset, count))
    offset += count
  end
end

# File lib/object_framework/http2/connection.rb, line 71
def send_headers(stream, headers, end_stream: false)
  stream.context.cancellation.check! unless end_stream
  block = @encoder.encode(headers)
  raise CapacityExceeded, "Outgoing headers exceed configured limit" if headers.sum { |name, value| name.bytesize + value.bytesize + 32 } > @header_limit
  @write_mutex.synchronize do
    offset, first = 0, true
    begin
      chunk = block.byteslice(offset, @peer_frame_size) || "".b
      offset += chunk.bytesize
      flags = offset == block.bytesize ? 4 : 0
      flags |= 1 if first && end_stream
      write_frame(first ? 1 : 9, flags, stream.id, chunk)
      first = false
    end while offset < block.bytesize
  end
  finish_local(stream) if end_stream
end