Class: Raptor::Http2
- Inherits:
-
Object
- Object
- Raptor::Http2
- Defined in:
- lib/raptor/http2.rb,
sig/generated/raptor/http2.rbs
Overview
Handles HTTP/2 request processing and Rack application integration.
Http2 manages the HTTP/2 protocol lifecycle including frame processing, HPACK header compression, stream management, and response writing. It integrates with the same reactor, ractor pool, and thread pool pipeline used by HTTP/1.1 connections.
Defined Under Namespace
Classes: FlowControl, Writer
Constant Summary collapse
- EAGER_READ_TIMEOUT =
0.001- EAGER_READ_BUFFER_SIZE =
64 * 1024
- EAGER_MAX_ROUNDS =
4- FLAG_END_STREAM =
0x1- FLAG_END_HEADERS =
0x4- FLAG_ACK =
0x1- FLAG_PRIORITY =
0x20- ERROR_NO_ERROR =
0x0- ERROR_PROTOCOL_ERROR =
0x1- DEFAULT_WINDOW_SIZE =
65_535- MAX_FRAME_SIZE =
16_384- SERVER_PROTOCOL =
"HTTP/2"- RACK_HEADER_PREFIX =
"rack."- HOP_BY_HOP_HEADERS =
["connection", "transfer-encoding", "keep-alive", "upgrade", "proxy-connection"].freeze
Instance Attribute Summary collapse
-
#initial_settings_frame ⇒ String
readonly
The initial server SETTINGS frame sent on every new HTTP/2 connection.
Class Method Summary collapse
-
.process_frames(data) ⇒ Hash
Processes HTTP/2 frames from the connection buffer.
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, 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_read_next_batch(socket) ⇒ String?
Reads the next frame batch from
socketwithin a short window, or returns nil if nothing arrives in time. -
#handle_parsed_request(result, reactor, thread_pool) ⇒ void
Handles a parsed HTTP/2 request from the ractor pool.
-
#initialize(app, server_port, connection_options: {}, http2_options: {}, access_log_io: nil, on_error: nil) ⇒ Http2
constructor
Creates a new Http2 handler.
-
#populate_server_name_and_port(env) ⇒ void
Populates SERVER_NAME and SERVER_PORT from the HTTP_HOST header.
-
#write_access_log(env, status, size, remote_addr) ⇒ void
Instance-level wrapper around Raptor::Http.write_access_log that routes to the configured
@access_log_io. -
#write_http2_error_response(socket, writer, stream_id) ⇒ void
Writes a 500 error response as HTTP/2 frames.
-
#write_http2_response(socket, writer, flow_control, stream_id, status, headers, body) ⇒ String
Writes a Rack response as HTTP/2 frames to the socket.
Constructor Details
#initialize(app, server_port, connection_options: {}, http2_options: {}, access_log_io: nil, on_error: nil) ⇒ Http2
Creates a new Http2 handler.
275 276 277 278 279 280 281 282 283 284 285 286 287 288 |
# File 'lib/raptor/http2.rb', line 275 def initialize(app, server_port, connection_options: {}, http2_options: {}, access_log_io: nil, on_error: nil) @app = app @server_port = server_port @write_timeout = [:write_timeout] || Http::WRITE_TIMEOUT @access_log_io = access_log_io @on_error = on_error parser = Http2Parser.new settings_payload = parser.build_settings( max_concurrent_streams: [:max_concurrent_streams], initial_window_size: DEFAULT_WINDOW_SIZE ) @initial_settings_frame = parser.build_frame(:settings, 0, 0, settings_payload).freeze end |
Instance Attribute Details
#initial_settings_frame ⇒ String (readonly)
The initial server SETTINGS frame sent on every new HTTP/2 connection.
260 261 262 |
# File 'lib/raptor/http2.rb', line 260 def initial_settings_frame @initial_settings_frame end |
Class Method Details
.process_frames(data) ⇒ Hash
Processes HTTP/2 frames from the connection buffer.
Parses frames, handles HPACK decoding, tracks stream state, and returns updated connection state along with any outgoing protocol frames and completed stream requests. Ractor-safe.
309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 |
# File 'lib/raptor/http2.rb', line 309 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 = [] window_updates = [] 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, window_updates, peer_initial_window_size, connection_window, preface_received, last_client_stream_id, pending_headers, false) end end loop do 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] & FLAG_ACK).zero? 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 = frame[:payload] unless streams.key?(stream_id) 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] & FLAG_PRIORITY) != 0 header_payload = header_payload.byteslice(5..-1) || "" end end_stream = (frame[:flags] & FLAG_END_STREAM) != 0 if (frame[:flags] & FLAG_END_HEADERS) != 0 decoded_headers, hpack_table = parser.parse_headers(header_payload, hpack_table) streams, completed_requests = finalize_headers(streams, completed_requests, stream_id, decoded_headers, end_stream) else pending_headers = { stream_id: stream_id, buffer: header_payload, end_stream: end_stream } end when :continuation if pending_headers.nil? || 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 (frame[:flags] & FLAG_END_HEADERS) != 0 stream_id = pending_headers[:stream_id] decoded_headers, hpack_table = parser.parse_headers(pending_headers[:buffer], hpack_table) streams, completed_requests = finalize_headers(streams, completed_requests, stream_id, decoded_headers, pending_headers[:end_stream]) pending_headers = nil end when :data stream_id = frame[:stream_id] unless streams.key?(stream_id) goaway_error = ERROR_PROTOCOL_ERROR break end stream = streams[stream_id] existing_body = stream[:body] || "" stream = stream.merge(body: existing_body + frame[:payload]) if frame[:payload].bytesize.positive? connection_window -= frame[:payload].bytesize if connection_window < DEFAULT_WINDOW_SIZE / 2 increment = DEFAULT_WINDOW_SIZE - connection_window wu_payload = [increment].pack("N") outgoing_frames << parser.build_frame(:window_update, 0, 0, wu_payload) outgoing_frames << parser.build_frame(:window_update, 0, stream_id, wu_payload) connection_window += increment end end if (frame[:flags] & FLAG_END_STREAM) != 0 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] & FLAG_ACK).zero? outgoing_frames << parser.build_frame(:ping, FLAG_ACK, 0, frame[:payload]) end when :goaway break when :rst_stream streams.delete(frame[:stream_id]) end 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, window_updates, peer_initial_window_size, connection_window, preface_received, last_client_stream_id, pending_headers, !goaway_error.nil?) 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.
594 595 596 597 598 599 600 601 602 603 604 605 606 |
# File 'lib/raptor/http2.rb', line 594 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, remote_addr:) ⇒ Hash
Builds a Rack environment hash from HTTP/2 headers and body.
Translates HTTP/2 pseudo-headers into Rack-compatible environment keys and populates all required Rack env entries.
774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 |
# File 'lib/raptor/http2.rb', line 774 def build_rack_env(headers, body, 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("-", "_")}" env[rack_key] = value 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_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 populate_server_name_and_port(env) env end |
#create_writer ⇒ Writer
Creates a per-connection Writer configured with the handler's write timeout.
295 296 297 |
# File 'lib/raptor/http2.rb', line 295 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.
647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 |
# File 'lib/raptor/http2.rb', line 647 def dispatch_stream_request(socket, writer, flow_control, stream_id, headers, body, remote_addr:) env = build_rack_env(headers, body, remote_addr: remote_addr) status, response_headers, response_body = @app.call(env) response_size = write_http2_response(socket, writer, flow_control, stream_id, status, response_headers, response_body) write_access_log(env, status, response_size, remote_addr) if @access_log_io rescue => error write_http2_error_response(socket, writer, stream_id) if @on_error @on_error.call(env, error) rescue nil else raise end ensure response_body.close if response_body.respond_to?(:close) flow_control.discard_stream(stream_id) if flow_control end |
#eager_read_next_batch(socket) ⇒ String?
Reads the next frame batch from socket within a short window, or
returns nil if nothing arrives in time.
615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 |
# File 'lib/raptor/http2.rb', line 615 def eager_read_next_batch(socket) return unless socket.wait_readable(EAGER_READ_TIMEOUT) data = begin socket.read_nonblock(EAGER_READ_BUFFER_SIZE) rescue IO::WaitReadable, EOFError, IOError return end buffer = String.new buffer << data while socket.pending > 0 buffer << socket.read_nonblock(socket.pending) end buffer end |
#handle_parsed_request(result, reactor, thread_pool) ⇒ void
This method returns an undefined value.
Handles a parsed HTTP/2 request from the ractor pool.
Writes outgoing protocol frames to the socket, updates reactor state, and dispatches completed stream requests to the thread pool. Eagerly consumes subsequent frame batches that are already buffered, skipping the reactor and ractor pool hops while the connection is hot.
539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 |
# File 'lib/raptor/http2.rb', line 539 def handle_parsed_request(result, reactor, thread_pool) socket = reactor.socket_for(result[:id]) return unless socket writer = reactor.writer_for(result[:id]) flow_control = reactor.flow_control_for(result[:id]) rounds = 0 loop do if flow_control && (result[:window_updates] || result[:peer_initial_window_size]) apply_flow_control_updates(flow_control, result) end writer.write_frames(socket, result[:outgoing_frames]) if result[:close_connection] reactor.close_connection(result[:id]) return end result[:completed_requests]&.each do |request| stream_id = request[:stream_id] remote_addr = result[:remote_addr] || Server::DEFAULT_REMOTE_ADDR 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 next_batch = eager_read_next_batch(socket) break unless next_batch result = Raptor::Http2.process_frames(result.merge(buffer: result[:buffer] + next_batch)) end reactor.update_http2_state(result) 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.
828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 |
# File 'lib/raptor/http2.rb', line 828 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 env[Rack::SERVER_PORT] ||= port || @server_port.to_s else env[Rack::SERVER_NAME] ||= Server::DEFAULT_SERVER_NAME env[Rack::SERVER_PORT] ||= @server_port.to_s end end |
#write_access_log(env, status, size, remote_addr) ⇒ void
This method returns an undefined value.
Instance-level wrapper around Raptor::Http.write_access_log that routes to
the configured @access_log_io.
759 760 761 |
# File 'lib/raptor/http2.rb', line 759 def write_access_log(env, status, size, remote_addr) Http.write_access_log(@access_log_io, env, status, size, remote_addr) end |
#write_http2_error_response(socket, writer, stream_id) ⇒ void
This method returns an undefined value.
Writes a 500 error response as HTTP/2 frames.
739 740 741 742 743 744 745 746 747 |
# File 'lib/raptor/http2.rb', line 739 def write_http2_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_http2_response(socket, writer, flow_control, stream_id, status, headers, body) ⇒ String
Writes a Rack response as HTTP/2 frames to the socket.
DATA frames are partitioned through flow_control so each write fits
within the peer's per-stream and connection windows.
681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 |
# File 'lib/raptor/http2.rb', line 681 def write_http2_response(socket, writer, flow_control, stream_id, status, headers, body) parser = Http2Parser.new header_pairs = [[":status", status.to_s]] headers.each do |name, value| lowered = name.downcase next if lowered.start_with?(RACK_HEADER_PREFIX) next if HOP_BY_HOP_HEADERS.include?(lowered) if value.is_a?(Array) value.each { |val| header_pairs << [lowered, val.to_s] } else header_pairs << [lowered, value.to_s] end end encoded_headers = parser.encode_headers(header_pairs) body_chunks = [] body_bytes = 0 body.each do |chunk| next if chunk.empty? body_chunks << chunk body_bytes += chunk.bytesize end if body_chunks.empty? writer.write_frames(socket, [parser.build_frame(:headers, FLAG_END_STREAM | FLAG_END_HEADERS, stream_id, encoded_headers)]) return "0" end frames = [parser.build_frame(:headers, FLAG_END_HEADERS, stream_id, encoded_headers)] last_chunk_index = body_chunks.size - 1 body_chunks.each_with_index do |chunk, chunk_index| offset = 0 while offset < chunk.bytesize remaining = chunk.bytesize - offset last_frame = chunk_index == last_chunk_index && remaining <= MAX_FRAME_SIZE granted = flow_control.acquire(stream_id, remaining, end_stream: last_frame) slice = offset == 0 && granted == chunk.bytesize ? chunk : chunk.byteslice(offset, granted) offset += granted end_stream = chunk_index == last_chunk_index && offset == chunk.bytesize frames << parser.build_frame(:data, end_stream ? FLAG_END_STREAM : 0, stream_id, slice) end end writer.write_frames(socket, frames) body_bytes.to_s end |