Class: Raptor::Http2

Inherits:

Object

Object
Raptor::Http2

show all

Defined in:: lib/raptor/http2.rb,
sig/generated/raptor/http2.rbs

Overview

Handles HTTP/2 request processing and Rack application integration.

Defined Under Namespace

Classes: FlowControl, ResponseStream, StreamClosedError, Writer

Constant Summary collapse

EAGER_READ_BUFFER_SIZE = Returns: (Object)

64 * 1024

EAGER_MAX_ROUNDS = Returns: (::Integer)

FLAG_END_STREAM = Returns: (::Integer)

0x1

FLAG_ACK = Returns: (::Integer)

0x1

FLAG_END_HEADERS = Returns: (::Integer)

0x4

FLAG_PADDED = Returns: (::Integer)

0x8

FLAG_PRIORITY = Returns: (::Integer)

0x20

ERROR_NO_ERROR = Returns: (::Integer)

0x0

ERROR_PROTOCOL_ERROR = Returns: (::Integer)

0x1

ERROR_INTERNAL_ERROR = Returns: (::Integer)

0x2

ERROR_FRAME_SIZE_ERROR = Returns: (::Integer)

0x6

ERROR_REFUSED_STREAM = Returns: (::Integer)

0x7

ERROR_COMPRESSION_ERROR = Returns: (::Integer)

0x9

DEFAULT_WINDOW_SIZE = Returns: (::Integer)

65_535

MAX_FRAME_SIZE = Returns: (::Integer)

16_384

RESPONSE_TRAILERS = Returns: (::String)

"raptor.response_trailers"

SERVER_PROTOCOL = Returns: (::String)

"HTTP/2"

REQUEST_PSEUDO_HEADERS = Returns: (Object)

[":method", ":scheme", ":path", ":authority"].freeze

REQUIRED_REQUEST_PSEUDO_HEADERS = Returns: (Object)

[":method", ":scheme", ":path"].freeze

Class Method Summary collapse

.process_frames(data) ⇒ Hash
Advances HTTP/2 frame parsing from the connection buffer, returning updated connection state along with any outgoing protocol frames and completed stream requests.

Instance Method Summary collapse

#apply_flow_control_updates(flow_control, result) ⇒ void
Applies inbound flow-control updates from a parsed result to the connection's FlowControl.
#build_rack_env(headers, body, socket, writer, flow_control, stream_id, remote_addr:) ⇒ Hash
Builds a Rack environment hash from HTTP/2 headers and body.
#create_writer ⇒ Writer
Creates a per-connection Writer configured with the handler's write timeout.
#dispatch_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:) ⇒ void
Dispatches a completed stream request to the Rack app and writes the response back as HTTP/2 frames.
#eager_accept(socket, id, reactor, thread_pool, remote_addr, url_scheme) ⇒ void
Sends the server SETTINGS frame on a freshly negotiated HTTP/2 connection, then eagerly reads and parses the first client frame batch, dispatching completed streams directly to the thread pool.
#eager_read_next_batch(socket) ⇒ String^?
Reads the next frame batch from socket if it has already arrived, without waiting on the shared collector thread.
#handle_parsed_request(parsed_request, reactor, thread_pool) ⇒ void
Handles a parsed HTTP/2 result.
#initialize(app, server_port, connection_options: {}, http2_options: {}, access_log_io: nil, clean_thread_locals: true, clean_fiber_locals: false, on_error: nil) ⇒ Http2 constructor
Creates a new Http2 handler.
#parser_worker ⇒ Proc
Returns a Ractor-safe proc that parses HTTP/2 frames from the state hash's buffered bytes.
#perform_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:) ⇒ void
Calls the Rack app and writes its response for one stream.
#populate_server_name_and_port(env) ⇒ void
Populates SERVER_NAME and SERVER_PORT from the HTTP_HOST header.
#send_early_hints(socket, writer, flow_control, stream_id, hints) ⇒ void
Sends an HTTP 103 Early Hints response.
#shutdown(reactor) ⇒ void
Stops new stream dispatch and asks clients to finish streams already handed to the application.
#write_access_log(env, status, size, remote_addr) ⇒ void
Writes a completed request to the access log.
#write_detached_response(socket, writer, flow_control, stream_id, status, headers, body, finished) { ... } ⇒ void
Starts a detached HTTP/2 response after writing its response headers.
#write_error_response(socket, writer, stream_id) ⇒ void
Writes a 500 error response as HTTP/2 frames.
#write_reset_stream(socket, writer, stream_id, error_code) ⇒ void
Resets one HTTP/2 stream with the given error code.
#write_response(socket, writer, flow_control, stream_id, status, headers, body, trailers:, request_method:) { ... } ⇒ String
Writes a Rack response as HTTP/2 frames to the socket, partitioning DATA frames through flow_control to fit within the peer's windows.

Constructor Details

#initialize(app, server_port, connection_options: {}, http2_options: {}, access_log_io: nil, clean_thread_locals: true, clean_fiber_locals: false, on_error: nil) ⇒ `Http2`

Creates a new Http2 handler.

Parameters:

app (#call) —
the Rack application to dispatch requests to
server_port (Integer) —
port number used to populate SERVER_PORT in the Rack env
connection_options (Hash) (defaults to: {}) —
per-connection settings shared across protocols
http2_options (Hash) (defaults to: {}) —
HTTP/2-specific settings
access_log_io (IO, nil) (defaults to: nil) —
IO to write Common Log Format access entries to, or nil to disable
clean_thread_locals (Boolean) (defaults to: true) —
whether to clear application thread locals after each request
clean_fiber_locals (Boolean) (defaults to: false) —
whether to process each request in a fresh Fiber
on_error (#call, nil) (defaults to: nil) —
callback invoked with (env, exception) when the Rack app raises
connection_options: (Hash[Symbol, untyped]) (defaults to: {})
http2_options: (Hash[Symbol, untyped]) (defaults to: {})
access_log_io: (IO, nil) (defaults to: nil)
clean_thread_locals: (Boolean) (defaults to: true)
clean_fiber_locals: (Boolean) (defaults to: false)
on_error: (^(Hash[String, untyped]?, Exception) -> void, nil) (defaults to: nil)

Options Hash (connection_options:):

:write_timeout (Integer) —
per-write socket timeout in seconds

Options Hash (http2_options:):

:max_concurrent_streams (Integer) —
maximum HTTP/2 concurrent streams per connection

# File 'lib/raptor/http2.rb', line 831

def initialize(app, server_port, connection_options: {}, http2_options: {}, access_log_io: nil, clean_thread_locals: true, clean_fiber_locals: false, on_error: nil)
  @app = app
  @server_port_string = server_port.to_s.freeze
  @write_timeout = connection_options[:write_timeout] || Http::WRITE_TIMEOUT
  @access_log_io = access_log_io
  @clean_thread_locals = clean_thread_locals
  @clean_fiber_locals = clean_fiber_locals
  @on_error = on_error
  @running = AtomicBoolean.new(true)

  parser = Http2Parser.new
  settings_payload = parser.build_settings(
    max_concurrent_streams: http2_options[:max_concurrent_streams],
    initial_window_size: DEFAULT_WINDOW_SIZE,
    max_header_list_size: Http2Parser::MAX_HEADER_LIST_SIZE
  )
  @initial_settings_frame = parser.build_frame(:settings, 0, 0, settings_payload).freeze
end

Class Method Details

.process_frames(data) ⇒ `Hash`

Advances HTTP/2 frame parsing from the connection buffer, returning updated connection state along with any outgoing protocol frames and completed stream requests.

Parameters:

data (Hash) —
the connection state including buffer and HPACK table

Returns:

(Hash) —
updated state with outgoing_frames and completed_requests

# File 'lib/raptor/http2.rb', line 462

def self.process_frames(data)
  parser = Http2Parser.new
  buffer = data[:buffer]
  hpack_table = data[:hpack_table] || []
  streams = data[:http2_streams] ? data[:http2_streams].dup : {}
  outgoing_frames = []
  completed_requests = []
  cancelled_streams = []
  window_updates = []
  ping_acknowledgements = []
  peer_initial_window_size = nil
  connection_window = data[:http2_window] || DEFAULT_WINDOW_SIZE
  preface_received = data[:http2_preface_received] || false
  last_client_stream_id = data[:http2_last_client_stream_id] || 0
  pending_headers = data[:http2_pending_headers]
  goaway_error = nil

  unless preface_received
    if buffer.bytesize >= 24 && buffer.byteslice(0, 24) == Http2Parser.connection_preface
      buffer = buffer.byteslice(24..-1) || ""
      preface_received = true
    else
      return build_result(data, buffer, hpack_table, streams, outgoing_frames, completed_requests, cancelled_streams, window_updates, ping_acknowledgements, peer_initial_window_size, connection_window, preface_received, last_client_stream_id, pending_headers, false)
    end
  end

  loop do
    # Reject oversized frames before waiting for their payload.
    if buffer.bytesize >= 3 && (buffer.getbyte(0) << 16 | buffer.getbyte(1) << 8 | buffer.getbyte(2)) > MAX_FRAME_SIZE
      goaway_error = ERROR_FRAME_SIZE_ERROR
      break
    end

    parsed = parser.parse_frame(buffer)
    break unless parsed

    frame, consumed = parsed
    buffer = buffer.byteslice(consumed..-1) || ""

    if pending_headers && frame[:type] != :continuation
      goaway_error = ERROR_PROTOCOL_ERROR
      break
    end

    case frame[:type]
    when :settings
      if frame[:flags].nobits?(FLAG_ACK)
        parsed_settings = parser.parse_settings(frame[:payload])
        peer_initial_window_size = parsed_settings[:initial_window_size] if parsed_settings.key?(:initial_window_size)
        outgoing_frames << parser.build_frame(:settings, FLAG_ACK, 0, nil)
      end

    when :headers
      stream_id = frame[:stream_id]
      header_payload = unpadded_payload(frame)
      trailers = streams.key?(stream_id)

      unless header_payload
        goaway_error = ERROR_PROTOCOL_ERROR
        break
      end

      unless trailers
        if stream_id.even? || stream_id <= last_client_stream_id
          goaway_error = ERROR_PROTOCOL_ERROR
          break
        end
        last_client_stream_id = stream_id
      end

      if frame[:flags].anybits?(FLAG_PRIORITY)
        header_payload = header_payload.byteslice(5..-1) || ""
      end

      end_stream = frame[:flags].anybits?(FLAG_END_STREAM)

      if frame[:flags].anybits?(FLAG_END_HEADERS)
        decoded_headers, hpack_table = parser.parse_headers(header_payload, hpack_table)
        streams, completed_requests = complete_header_block(
          streams, completed_requests, outgoing_frames, parser,
          stream_id, decoded_headers, end_stream, trailers
        )
      else
        pending_headers = { stream_id: stream_id, buffer: header_payload, end_stream: end_stream, trailers: trailers }
      end

    when :continuation
      if !pending_headers || frame[:stream_id] != pending_headers[:stream_id]
        goaway_error = ERROR_PROTOCOL_ERROR
        break
      end

      pending_headers = pending_headers.merge(buffer: pending_headers[:buffer] + frame[:payload])
      if pending_headers[:buffer].bytesize > Http2Parser::MAX_HEADER_LIST_SIZE
        goaway_error = ERROR_PROTOCOL_ERROR
        break
      end

      if frame[:flags].anybits?(FLAG_END_HEADERS)
        stream_id = pending_headers[:stream_id]
        decoded_headers, hpack_table = parser.parse_headers(pending_headers[:buffer], hpack_table)
        streams, completed_requests = complete_header_block(
          streams, completed_requests, outgoing_frames, parser,
          stream_id, decoded_headers, pending_headers[:end_stream], pending_headers[:trailers]
        )
        pending_headers = nil
      end

    when :data
      stream_id = frame[:stream_id]
      chunk = unpadded_payload(frame)

      if stream_id.zero? || stream_id > last_client_stream_id || !chunk
        goaway_error = ERROR_PROTOCOL_ERROR
        break
      end

      # DATA for closed streams still counts toward connection flow control.
      connection_window -= frame[:length]
      if connection_window < DEFAULT_WINDOW_SIZE / 2
        outgoing_frames << parser.build_frame(:window_update, 0, 0, [DEFAULT_WINDOW_SIZE - connection_window].pack("N"))
        connection_window = DEFAULT_WINDOW_SIZE
      end

      stream = streams[stream_id]
      next unless stream

      stream_window_used = (stream[:window_used] || 0) + frame[:length]
      if stream_window_used >= DEFAULT_WINDOW_SIZE / 2 && frame[:flags].nobits?(FLAG_END_STREAM)
        outgoing_frames << parser.build_frame(:window_update, 0, stream_id, [stream_window_used].pack("N"))
        stream_window_used = 0
      end
      stream = stream.merge(body: (stream[:body] || "") + chunk, window_used: stream_window_used)

      if frame[:flags].anybits?(FLAG_END_STREAM)
        stream_headers = stream[:headers] || []
        completed_requests << {
          stream_id: stream_id,
          headers: stream_headers,
          body: stream[:body]
        }

        streams.delete(stream_id)
      else
        streams[stream_id] = stream
      end

    when :window_update
      increment = parser.parse_window_update(frame[:payload])
      window_updates << [frame[:stream_id], increment]

    when :ping
      if frame[:flags].anybits?(FLAG_ACK)
        ping_acknowledgements << frame[:payload]
      else
        outgoing_frames << parser.build_frame(:ping, FLAG_ACK, 0, frame[:payload])
      end

    when :goaway
      break

    when :rst_stream
      stream_id = frame[:stream_id]
      streams.delete(stream_id)
      completed_requests.reject! { |request| request[:stream_id] == stream_id }
      cancelled_streams << stream_id
    end
  rescue Http2ParserError
    # A failed header decode can desynchronize HPACK state for the whole connection.
    goaway_error = if frame[:type] == :headers || frame[:type] == :continuation
      ERROR_COMPRESSION_ERROR
    else
      ERROR_FRAME_SIZE_ERROR
    end
    break
  end

  if goaway_error
    goaway_payload = [last_client_stream_id, goaway_error].pack("NN")
    outgoing_frames << parser.build_frame(:goaway, 0, 0, goaway_payload)
  end

  build_result(data, buffer, hpack_table, streams, outgoing_frames, completed_requests, cancelled_streams, window_updates, ping_acknowledgements, peer_initial_window_size, connection_window, preface_received, last_client_stream_id, pending_headers, !!goaway_error)
end

Instance Method Details

#apply_flow_control_updates(flow_control, result) ⇒ `void`

This method returns an undefined value.

Applies inbound flow-control updates from a parsed result to the connection's FlowControl.

Parameters:

flow_control (FlowControl) —
per-connection outbound flow controller
result (Hash) —
the parsed result from process_frames

# File 'lib/raptor/http2.rb', line 1034

def apply_flow_control_updates(flow_control, result)
  result[:window_updates]&.each do |stream_id, increment|
    if stream_id.zero?
      flow_control.add_connection_window(increment)
    else
      flow_control.add_stream_window(stream_id, increment)
    end
  end

  if (new_size = result[:peer_initial_window_size])
    flow_control.set_initial_stream_window(new_size)
  end
end

#build_rack_env(headers, body, socket, writer, flow_control, stream_id, remote_addr:) ⇒ `Hash`

Builds a Rack environment hash from HTTP/2 headers and body.

Parameters:

headers (Array<Array(String, String)>) —
HTTP/2 header pairs
body (String) —
the request body
socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket
writer (Writer) —
frame writer for the connection
flow_control (FlowControl) —
per-connection outbound flow controller
stream_id (Integer) —
the HTTP/2 stream identifier
remote_addr (String) —
the client IP address
remote_addr: (String)

Returns:

(Hash) —
fully populated Rack environment hash

# File 'lib/raptor/http2.rb', line 1299

def build_rack_env(headers, body, socket, writer, flow_control, stream_id, remote_addr:)
  env = {}

  headers.each do |name, value|
    if name.start_with?(":")
      case name
      when ":method"    then env[Rack::REQUEST_METHOD] = value
      when ":path"
        path, query = value.split("?", 2)
        env[Rack::PATH_INFO] = path
        env[Rack::QUERY_STRING] = query || ""
      when ":scheme"    then env[Rack::RACK_URL_SCHEME] = value
      when ":authority" then env[Rack::HTTP_HOST] = value
      end
    elsif name == "content-type"
      env[Http::CONTENT_TYPE] = value
    elsif name == "content-length"
      env[Http::CONTENT_LENGTH] = value
    else
      rack_key = "HTTP_#{name.upcase.tr("-", "_")}"
      if (existing = env[rack_key])
        separator = rack_key == Rack::HTTP_COOKIE ? "; " : ", "
        env[rack_key] = "#{existing}#{separator}#{value}"
      else
        env[rack_key] = value
      end
    end
  end

  env[Rack::SERVER_PROTOCOL] = SERVER_PROTOCOL
  env[Rack::RACK_VERSION] = Rack::VERSION
  env[Rack::RACK_INPUT] = StringIO.new(body).set_encoding(Encoding::ASCII_8BIT)
  env[Rack::RACK_ERRORS] = $stderr
  env[Rack::RACK_RESPONSE_FINISHED] = []
  env[Rack::RACK_EARLY_HINTS] = proc do |hints|
    send_early_hints(socket, writer, flow_control, stream_id, hints) rescue nil
  end
  env[Rack::RACK_IS_HIJACK] = false

  env[Rack::SCRIPT_NAME] = "" unless env.key?(Rack::SCRIPT_NAME)
  env[Rack::PATH_INFO] = "" unless env.key?(Rack::PATH_INFO)
  env[Rack::QUERY_STRING] = "" unless env.key?(Rack::QUERY_STRING)

  if body.bytesize.positive? && !env.key?(Http::CONTENT_LENGTH)
    env[Http::CONTENT_LENGTH] = body.bytesize.to_s
  end

  env[Http::REMOTE_ADDR] = remote_addr
  env[Http::SERVER_SOFTWARE] = Http::SERVER_SOFTWARE_VALUE
  env[Http::HTTP_VERSION] = SERVER_PROTOCOL
  env[RESPONSE_TRAILERS] = {}

  populate_server_name_and_port(env)

  env
end

#create_writer ⇒ `Writer`

Creates a per-connection Writer configured with the handler's write timeout.

Returns:

(Writer) —
a new per-connection frame writer



1022
1023
1024

# File 'lib/raptor/http2.rb', line 1022

def create_writer
  Writer.new(write_timeout: @write_timeout)
end

#dispatch_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:) ⇒ `void`

This method returns an undefined value.

Dispatches a completed stream request to the Rack app and writes the response back as HTTP/2 frames.

Parameters:

socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket
writer (Writer) —
frame writer for the connection
flow_control (FlowControl) —
per-connection outbound flow controller
stream_id (Integer) —
the HTTP/2 stream identifier
headers (Array<Array(String, String)>) —
request headers
body (String) —
request body
remote_addr (String) —
the client IP address
remote_addr: (String)

# File 'lib/raptor/http2.rb', line 1084

def dispatch_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:)
  if @clean_fiber_locals
    Fiber.new { perform_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr: remote_addr) }.resume
  else
    perform_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr: remote_addr)
  end
ensure
  ThreadLocals.clear if @clean_thread_locals
end

#eager_accept(socket, id, reactor, thread_pool, remote_addr, url_scheme) ⇒ `void`

This method returns an undefined value.

Sends the server SETTINGS frame on a freshly negotiated HTTP/2 connection, then eagerly reads and parses the first client frame batch, dispatching completed streams directly to the thread pool. Falls back to the reactor when no initial data is ready.

Parameters:

socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket
id (Integer) —
unique connection identifier
reactor (Reactor) —
the reactor managing the connection
thread_pool (AtomicThreadPool) —
thread pool for application processing
remote_addr (String) —
the client IP address
url_scheme (String) —
"http" or "https"

# File 'lib/raptor/http2.rb', line 892

def eager_accept(socket, id, reactor, thread_pool, remote_addr, url_scheme)
  unless @running.true?
    socket.close rescue nil
    return
  end

  writer = create_writer
  flow_control = FlowControl.new
  ping_payload = [id].pack("Q>")
  ping_frame = Http2Parser.new.build_frame(:ping, 0, 0, ping_payload).freeze
  initial_state = {
    id: id,
    protocol: :http2,
    remote_addr: remote_addr,
    url_scheme: url_scheme
  }

  reactor.attach_http2(
    id: id,
    socket: socket,
    state: initial_state,
    writer: writer,
    flow_control: flow_control,
    ping_frame: ping_frame,
    ping_payload: ping_payload
  )
  unless @running.true?
    reactor.close_connection(id)
    return
  end

  socket.write(@initial_settings_frame) rescue nil

  buffer = begin
    socket.read_nonblock(EAGER_READ_BUFFER_SIZE)
  rescue IO::WaitReadable
    reactor.watch(id)
    return
  rescue EOFError, IOError, SystemCallError
    reactor.close_connection(id)
    return
  end

  while socket.respond_to?(:pending) && socket.pending.positive?
    buffer << socket.read_nonblock(socket.pending)
  end

  result = Http2.process_frames(initial_state.merge(buffer: buffer))
  handle_parsed_request(result, reactor, thread_pool)
rescue => error
  Log.rescued_error(error)
  reactor.close_connection(id)
end

#eager_read_next_batch(socket) ⇒ `String`^?

Reads the next frame batch from socket if it has already arrived, without waiting on the shared collector thread.

Parameters:

socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket

Returns:

(String, nil) —
the bytes read, or nil if nothing was available

# File 'lib/raptor/http2.rb', line 1055

def eager_read_next_batch(socket)
  return unless socket.wait_readable(0)

  buffer = begin
    socket.read_nonblock(EAGER_READ_BUFFER_SIZE)
  rescue IO::WaitReadable, EOFError, IOError, SystemCallError
    return
  end

  while socket.respond_to?(:pending) && socket.pending.positive?
    buffer << socket.read_nonblock(socket.pending)
  end

  buffer
end

#handle_parsed_request(parsed_request, reactor, thread_pool) ⇒ `void`

This method returns an undefined value.

Handles a parsed HTTP/2 result. Writes outgoing frames, dispatches completed stream requests to the thread pool, and eagerly consumes further buffered frame batches before returning control to the reactor.

Parameters:

parsed_request (Hash) —
the parsed result produced by process_frames
reactor (Reactor) —
the reactor managing the connection
thread_pool (AtomicThreadPool) —
thread pool for application processing

# File 'lib/raptor/http2.rb', line 957

def handle_parsed_request(parsed_request, reactor, thread_pool)
  socket = reactor.socket_for(parsed_request[:id])
  return unless socket

  writer = reactor.writer_for(parsed_request[:id])
  flow_control = reactor.flow_control_for(parsed_request[:id])

  rounds = 0
  loop do
    reactor.acknowledge_http2_ping(parsed_request[:id], parsed_request[:ping_acknowledgements])
    parsed_request[:cancelled_streams]&.each do |stream_id|
      flow_control.cancel_stream(stream_id)
      reactor.cancel_http2_body(parsed_request[:id], stream_id)
    end

    if flow_control && (parsed_request[:window_updates]&.any? || parsed_request[:peer_initial_window_size])
      apply_flow_control_updates(flow_control, parsed_request)
      reactor.resume_http2_bodies(parsed_request[:id])
    end

    writer.write_frames(socket, parsed_request[:outgoing_frames])

    if parsed_request[:close_connection]
      reactor.close_connection(parsed_request[:id])
      return
    end

    parsed_request[:completed_requests]&.each do |request|
      stream_id = request[:stream_id]
      remote_addr = parsed_request[:remote_addr] || Server::DEFAULT_REMOTE_ADDR

      unless reactor.dispatch_http2_stream(parsed_request[:id], stream_id)
        write_reset_stream(socket, writer, stream_id, ERROR_REFUSED_STREAM)
        next
      end

      thread_pool << proc do
        dispatch_stream_request(
          socket, writer, flow_control, stream_id,
          request[:headers], request[:body],
          remote_addr: remote_addr
        )
      end
    end

    rounds += 1
    break if rounds >= EAGER_MAX_ROUNDS
    break if thread_pool.queue_size >= thread_pool.size

    next_batch = eager_read_next_batch(socket)
    break unless next_batch

    parsed_request = Http2.process_frames(parsed_request.merge(buffer: parsed_request[:buffer] + next_batch))
  end

  reactor.update_state(parsed_request)
end

#parser_worker ⇒ `Proc`

Returns a Ractor-safe proc that parses HTTP/2 frames from the state hash's buffered bytes.

Returns:

(Proc)

# File 'lib/raptor/http2.rb', line 856

def parser_worker
  proc do |data|
    Http2.process_frames(data)
  end
end

#perform_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:) ⇒ `void`

This method returns an undefined value.

Calls the Rack app and writes its response for one stream.

Parameters:

socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket
writer (Writer) —
frame writer for the connection
flow_control (FlowControl) —
per-connection outbound flow controller
stream_id (Integer) —
the HTTP/2 stream identifier
headers (Array<Array(String, String)>) —
request headers
body (String) —
request body
remote_addr (String) —
the client IP address
remote_addr: (String)

# File 'lib/raptor/http2.rb', line 1106

def perform_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:)
  env = nil
  status = nil
  response_headers = nil
  response_started = false
  detached = false

  flow_control.check(stream_id)
  env = build_rack_env(headers, body, socket, writer, flow_control, stream_id, remote_addr: remote_addr)
  status, response_headers, response_body = @app.call(env)

  no_body = env[Rack::REQUEST_METHOD] == "HEAD" || Http.no_entity_body_status?(status)
  if response_body.is_a?(DetachedBody) && !no_body
    finished = proc do |reason|
      error = StreamClosedError.new unless reason == :closed
      write_access_log(env, status, "-", remote_addr) if @access_log_io
      Http.call_response_finished(env, status, response_headers, error)
      flow_control.discard_stream(stream_id)
    end
    write_detached_response(socket, writer, flow_control, stream_id, status, response_headers, response_body, finished) do
      response_started = true
    end
    detached = true
    return
  end

  response_size = write_response(
    socket,
    writer,
    flow_control,
    stream_id,
    status,
    response_headers,
    response_body,
    trailers: env[RESPONSE_TRAILERS],
    request_method: env[Rack::REQUEST_METHOD]
  ) { response_started = true }
  if response_body.is_a?(DetachedBody)
    response_body.finish(:closed)
  elsif response_body.respond_to?(:close)
    response_body.close
  end
  write_access_log(env, status, response_size, remote_addr) if @access_log_io
  Http.call_response_finished(env, status, response_headers, nil)
rescue StreamClosedError => error
  Http.call_response_finished(env, status, response_headers, error)
rescue => error
  Http.call_response_finished(env, status, response_headers, error)
  if response_started
    write_reset_stream(socket, writer, stream_id, ERROR_INTERNAL_ERROR)
  else
    write_error_response(socket, writer, stream_id)
  end

  if @on_error
    @on_error.call(env, error) rescue nil
  else
    raise
  end
ensure
  flow_control.discard_stream(stream_id) if flow_control && !detached
end

#populate_server_name_and_port(env) ⇒ `void`

This method returns an undefined value.

Populates SERVER_NAME and SERVER_PORT from the HTTP_HOST header.

Parameters:

env (Hash) —
the Rack environment to populate

# File 'lib/raptor/http2.rb', line 1362

def populate_server_name_and_port(env)
  http_host = env[Rack::HTTP_HOST]

  if http_host
    if http_host.start_with?("[")
      host = http_host[/\A\[([^\]]+)\]/, 1]
      port = http_host[/\]:(\d+)\z/, 1]
    else
      host, port = http_host.split(":", 2)
    end
    env[Rack::SERVER_NAME] ||= host || Server::DEFAULT_SERVER_NAME
    env[Rack::SERVER_PORT] ||= port || @server_port_string
  else
    env[Rack::SERVER_NAME] ||= Server::DEFAULT_SERVER_NAME
    env[Rack::SERVER_PORT] ||= @server_port_string
  end
end

#send_early_hints(socket, writer, flow_control, stream_id, hints) ⇒ `void`

This method returns an undefined value.

Sends an HTTP 103 Early Hints response.

Parameters:

socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket
writer (Writer) —
frame writer for the connection
flow_control (FlowControl) —
per-connection outbound flow controller
stream_id (Integer) —
the HTTP/2 stream identifier
hints (Hash) —
response headers to send as early hints

# File 'lib/raptor/http2.rb', line 1245

def send_early_hints(socket, writer, flow_control, stream_id, hints)
  return if hints.empty?

  flow_control.check(stream_id)
  parser = Http2Parser.new
  encoded = parser.encode_response_headers(103, hints)
  writer.write_frames(socket, [parser.build_frame(:headers, FLAG_END_HEADERS, stream_id, encoded)])
end

#shutdown(reactor) ⇒ `void`

This method returns an undefined value.

Stops new stream dispatch and asks clients to finish streams already handed to the application.

Parameters:

reactor (Reactor) —
the reactor managing HTTP/2 connections

# File 'lib/raptor/http2.rb', line 869

def shutdown(reactor)
  @running.make_false
  parser = Http2Parser.new
  reactor.drain_http2 do |stream_id|
    payload = [stream_id, ERROR_NO_ERROR].pack("NN")
    parser.build_frame(:goaway, 0, 0, payload)
  end
end

#write_access_log(env, status, size, remote_addr) ⇒ `void`

This method returns an undefined value.

Writes a completed request to the access log.

Parameters:

env (Hash) —
the Rack environment
status (Integer) —
the response status code
size (String) —
the response body size in bytes, or - if unknown
remote_addr (String) —
the client IP address



1389
1390
1391

# File 'lib/raptor/http2.rb', line 1389

def write_access_log(env, status, size, remote_addr)
  Http.write_access_log(@access_log_io, env, status, size, remote_addr)
end

#write_detached_response(socket, writer, flow_control, stream_id, status, headers, body, finished) { ... } ⇒ `void`

This method returns an undefined value.

Starts a detached HTTP/2 response after writing its response headers.

Parameters:

socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket
writer (Writer) —
frame writer for the connection
flow_control (FlowControl) —
per-connection outbound flow controller
stream_id (Integer) —
the HTTP/2 stream identifier
status (Integer) —
HTTP status code
headers (Hash) —
response headers from the Rack application
body (DetachedBody) —
the response body
finished (Proc) —
called with the close reason

Yields:

once the response headers are queued

Yield Returns:

(void)

# File 'lib/raptor/http2.rb', line 1183

def write_detached_response(socket, writer, flow_control, stream_id, status, headers, body, finished)
  flow_control.check(stream_id)
  parser = Http2Parser.new
  encoded = parser.encode_response_headers(status, headers)
  writer.write_frames(socket, [parser.build_frame(:headers, FLAG_END_HEADERS, stream_id, encoded)])
  yield
  attached = writer.attach_body(stream_id, body, finished)
  write_reset_stream(socket, writer, stream_id, ERROR_REFUSED_STREAM) unless attached
end

#write_error_response(socket, writer, stream_id) ⇒ `void`

This method returns an undefined value.

Writes a 500 error response as HTTP/2 frames.

Parameters:

socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket
writer (Writer) —
frame writer for the connection
stream_id (Integer) —
the HTTP/2 stream identifier

# File 'lib/raptor/http2.rb', line 1277

def write_error_response(socket, writer, stream_id)
  parser = Http2Parser.new
  encoded = parser.encode_headers([[":status", "500"]])

  writer.write_frames(
    socket,
    [parser.build_frame(:headers, FLAG_END_STREAM | FLAG_END_HEADERS, stream_id, encoded)]
  )
end

#write_reset_stream(socket, writer, stream_id, error_code) ⇒ `void`

This method returns an undefined value.

Resets one HTTP/2 stream with the given error code.

Parameters:

socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket
writer (Writer) —
frame writer for the connection
stream_id (Integer) —
the HTTP/2 stream identifier
error_code (Integer) —
the HTTP/2 error code

# File 'lib/raptor/http2.rb', line 1263

def write_reset_stream(socket, writer, stream_id, error_code)
  parser = Http2Parser.new
  frame = parser.build_frame(:rst_stream, 0, stream_id, [error_code].pack("N"))
  writer.write_frames(socket, [frame])
end

#write_response(socket, writer, flow_control, stream_id, status, headers, body, trailers:, request_method:) { ... } ⇒ `String`

Writes a Rack response as HTTP/2 frames to the socket, partitioning DATA frames through flow_control to fit within the peer's windows.

Parameters:

socket (TCPSocket, OpenSSL::SSL::SSLSocket) —
the connection socket
writer (Writer) —
frame writer for the connection
flow_control (FlowControl) —
per-connection outbound flow controller
stream_id (Integer) —
the HTTP/2 stream identifier
status (Integer) —
HTTP status code
headers (Hash) —
response headers from the Rack application
body (Object) —
response body responding to each
trailers (Hash) —
trailing response headers populated by the Rack application
request_method (String) —
request method used to suppress HEAD response bodies
trailers: (Hash[String, String | Array[String]])
request_method: (String)

Yields:

once the response headers are queued

Returns:

(String) —
the response body size in bytes

# File 'lib/raptor/http2.rb', line 1209

def write_response(socket, writer, flow_control, stream_id, status, headers, body, trailers:, request_method:)
  parser = Http2Parser.new

  encoded_headers = parser.encode_response_headers(status, headers)
  flow_control.check(stream_id)
  no_body = request_method == "HEAD" || Http.no_entity_body_status?(status)
  if no_body
    writer.write_frames(socket, [parser.build_frame(:headers, FLAG_END_STREAM | FLAG_END_HEADERS, stream_id, encoded_headers)])
    yield if block_given?
    return "0"
  end

  writer.write_frames(socket, [parser.build_frame(:headers, FLAG_END_HEADERS, stream_id, encoded_headers)])
  yield if block_given?

  stream = ResponseStream.new(socket, writer, flow_control, stream_id, trailers)
  if body.respond_to?(:each)
    body.each { |chunk| stream.write(chunk) }
  else
    body.call(stream)
  end
  stream.close

  stream.bytes_written.to_s
end

Class: Raptor::Http2

Overview

Defined Under Namespace

Constant Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(app, server_port, connection_options: {}, http2_options: {}, access_log_io: nil, clean_thread_locals: true, clean_fiber_locals: false, on_error: nil) ⇒ Http2

Class Method Details

.process_frames(data) ⇒ Hash

Instance Method Details

#apply_flow_control_updates(flow_control, result) ⇒ void

#build_rack_env(headers, body, socket, writer, flow_control, stream_id, remote_addr:) ⇒ Hash

#create_writer ⇒ Writer

#dispatch_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:) ⇒ void

#eager_accept(socket, id, reactor, thread_pool, remote_addr, url_scheme) ⇒ void

#eager_read_next_batch(socket) ⇒ String?

#handle_parsed_request(parsed_request, reactor, thread_pool) ⇒ void

#parser_worker ⇒ Proc

#perform_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:) ⇒ void

#populate_server_name_and_port(env) ⇒ void

#send_early_hints(socket, writer, flow_control, stream_id, hints) ⇒ void

#shutdown(reactor) ⇒ void

#write_access_log(env, status, size, remote_addr) ⇒ void

#write_detached_response(socket, writer, flow_control, stream_id, status, headers, body, finished) { ... } ⇒ void

#write_error_response(socket, writer, stream_id) ⇒ void

#write_reset_stream(socket, writer, stream_id, error_code) ⇒ void

#write_response(socket, writer, flow_control, stream_id, status, headers, body, trailers:, request_method:) { ... } ⇒ String

#initialize(app, server_port, connection_options: {}, http2_options: {}, access_log_io: nil, clean_thread_locals: true, clean_fiber_locals: false, on_error: nil) ⇒ `Http2`

.process_frames(data) ⇒ `Hash`

#apply_flow_control_updates(flow_control, result) ⇒ `void`

#build_rack_env(headers, body, socket, writer, flow_control, stream_id, remote_addr:) ⇒ `Hash`

#create_writer ⇒ `Writer`

#dispatch_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:) ⇒ `void`

#eager_accept(socket, id, reactor, thread_pool, remote_addr, url_scheme) ⇒ `void`

#eager_read_next_batch(socket) ⇒ `String`^?

#handle_parsed_request(parsed_request, reactor, thread_pool) ⇒ `void`

#parser_worker ⇒ `Proc`

#perform_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:) ⇒ `void`

#populate_server_name_and_port(env) ⇒ `void`

#send_early_hints(socket, writer, flow_control, stream_id, hints) ⇒ `void`

#shutdown(reactor) ⇒ `void`

#write_access_log(env, status, size, remote_addr) ⇒ `void`

#write_detached_response(socket, writer, flow_control, stream_id, status, headers, body, finished) { ... } ⇒ `void`

#write_error_response(socket, writer, stream_id) ⇒ `void`

#write_reset_stream(socket, writer, stream_id, error_code) ⇒ `void`

#write_response(socket, writer, flow_control, stream_id, status, headers, body, trailers:, request_method:) { ... } ⇒ `String`