Class: Raptor::Http2

Inherits:
Object
  • Object
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

RESPONSE_TRAILERS =

Returns:

  • (::String) 
"raptor.response_trailers"
EAGER_READ_BUFFER_SIZE =

Returns:

  • (Object) 
64 * 1024
EAGER_MAX_ROUNDS =

Returns:

  • (::Integer) 
8
FLAG_END_STREAM =

Returns:

  • (::Integer) 
0x1
FLAG_END_HEADERS =

Returns:

  • (::Integer) 
0x4
FLAG_ACK =

Returns:

  • (::Integer) 
0x1
FLAG_PRIORITY =

Returns:

  • (::Integer) 
0x20
ERROR_NO_ERROR =

Returns:

  • (::Integer) 
0x0
ERROR_PROTOCOL_ERROR =

Returns:

  • (::Integer) 
0x1
ERROR_INTERNAL_ERROR =

Returns:

  • (::Integer) 
0x2
ERROR_REFUSED_STREAM =

Returns:

  • (::Integer) 
0x7
DEFAULT_WINDOW_SIZE =

Returns:

  • (::Integer) 
65_535
MAX_FRAME_SIZE =

Returns:

  • (::Integer) 
16_384
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

Instance Attribute 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

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



506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
 
# File 'lib/raptor/http2.rb', line 506

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 = server_port
  @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
  )
  @initial_settings_frame = parser.build_frame(:settings, 0, 0, settings_payload).freeze
end

Instance Attribute Details

#initial_settings_frame ⇒ String (readonly)

Returns the initial server SETTINGS frame to send on every new HTTP/2 connection.

Returns:

  • (String) 


489
490
491
 
# File 'lib/raptor/http2.rb', line 489

def initial_settings_frame
  @initial_settings_frame
end

Class Method Details

.invalid_pseudo_headers?(headers) ⇒ Boolean

Returns true when a decoded header block violates the HTTP/2 pseudo-header rules from RFC 9113 section 8.3: an unknown pseudo-header, a duplicate pseudo-header, a pseudo-header appearing after any regular header, or a required pseudo-header (:method, :scheme, :path) missing on non-CONNECT requests.

Parameters:

  • headers (Array<Array(String, String)>) —

    decoded header pairs

Returns:

  • (Boolean) 


455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
 
# File 'lib/raptor/http2.rb', line 455

def self.invalid_pseudo_headers?(headers)
  seen_pseudo = {}
  seen_regular = false

  headers.each do |name, _value|
    if name.start_with?(":")
      return true if seen_regular
      return true unless REQUEST_PSEUDO_HEADERS.include?(name)
      return true if seen_pseudo[name]
      seen_pseudo[name] = true
    else
      seen_regular = true
    end
  end

  return false if seen_pseudo[":method"] && headers.assoc(":method")&.last == "CONNECT"

  REQUIRED_REQUEST_PSEUDO_HEADERS.any? { |name| !seen_pseudo[name] }
end

.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



569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
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
 
# File 'lib/raptor/http2.rb', line 569

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
    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 = frame[:payload]
      trailers = streams.key?(stream_id)

      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 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]

      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].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
  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) —

    the per-connection flow controller

  • result (Hash) —

    the parsed result from process_frames



990
991
992
993
994
995
996
997
998
999
1000
1001
1002
 
# File 'lib/raptor/http2.rb', line 990

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 (OpenSSL::SSL::SSLSocket) —

    the connection socket

  • writer (Writer) —

    frame writer for the connection

  • stream_id (Integer) —

    the HTTP/2 stream identifier

  • remote_addr (String) —

    the client IP address

  • flow_control (FlowControl)
  • remote_addr: (String)

Returns:

  • (Hash) —

    fully populated Rack environment hash



1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
 
# File 'lib/raptor/http2.rb', line 1270

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("-", "_")}"
      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_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



529
530
531
 
# File 'lib/raptor/http2.rb', line 529

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 (OpenSSL::SSL::SSLSocket) —

    the connection socket

  • writer (Writer) —

    lock-free 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) 


1043
1044
1045
1046
1047
1048
1049
1050
1051
 
# File 'lib/raptor/http2.rb', line 1043

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:) }.resume
  else
    perform_stream_request(socket, writer, flow_control, stream_id, headers, body, 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 (OpenSSL::SSL::SSLSocket) —

    the connection socket

  • id (Integer) —

    unique client identifier

  • reactor (Reactor) —

    the reactor managing the connection

  • thread_pool (AtomicThreadPool) —

    thread pool for application processing

  • remote_addr (String) —

    client IP address

  • url_scheme (String) —

    "https"



857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
 
# File 'lib/raptor/http2.rb', line 857

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 = Raptor::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 (OpenSSL::SSL::SSLSocket) —

    the connection socket

Returns:

  • (String, nil) —

    the bytes read, or nil if nothing was available



1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
 
# File 'lib/raptor/http2.rb', line 1011

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

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

  buffer = String.new
  buffer << data

  while socket.respond_to?(:pending) && socket.pending.positive?
    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 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:

  • result (Hash) —

    the parsed result produced by process_frames

  • reactor (Reactor) —

    the reactor managing the connection

  • thread_pool (AtomicThreadPool) —

    thread pool for Rack app dispatch



922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
 
# File 'lib/raptor/http2.rb', line 922

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
    reactor.acknowledge_http2_ping(result[:id], result[:ping_acknowledgements])
    result[:cancelled_streams]&.each do |stream_id|
      flow_control.cancel_stream(stream_id)
      reactor.cancel_http2_body(result[:id], stream_id)
    end

    if flow_control && (result[:window_updates]&.any? || result[:peer_initial_window_size])
      apply_flow_control_updates(flow_control, result)
      reactor.resume_http2_bodies(result[:id])
    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

      unless reactor.dispatch_http2_stream(result[:id], stream_id)
        write_http2_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

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

  reactor.update_http2_state(result)
end

#parser_worker ⇒ Proc

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

Returns:

  • (Proc) 


555
556
557
558
559
 
# File 'lib/raptor/http2.rb', line 555

def parser_worker
  proc do |data|
    Raptor::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 (OpenSSL::SSL::SSLSocket) —

    the connection socket

  • writer (Writer) —

    lock-free 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) 


1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
 
# File 'lib/raptor/http2.rb', line 1065

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_http2_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_http2_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_http2_reset_stream(socket, writer, stream_id, ERROR_INTERNAL_ERROR)
  else
    write_http2_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



1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
 
# File 'lib/raptor/http2.rb', line 1328

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

#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 (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



1203
1204
1205
1206
1207
1208
1209
1210
 
# File 'lib/raptor/http2.rb', line 1203

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



540
541
542
543
544
545
546
547
 
# File 'lib/raptor/http2.rb', line 540

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.

Instance-level wrapper around Raptor::Http.write_access_log that routes to the configured @access_log_io.

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



1255
1256
1257
 
# File 'lib/raptor/http2.rb', line 1255

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

#write_http2_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 (OpenSSL::SSL::SSLSocket) —

    the connection socket

  • writer (Writer) —

    lock-free 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) 


1142
1143
1144
1145
1146
1147
1148
1149
1150
 
# File 'lib/raptor/http2.rb', line 1142

def write_http2_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_http2_reset_stream(socket, writer, stream_id, ERROR_REFUSED_STREAM) unless attached
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.

Parameters:

  • socket (OpenSSL::SSL::SSLSocket) —

    the connection socket

  • writer (Writer) —

    lock-free frame writer for the connection

  • stream_id (Integer) —

    the HTTP/2 stream identifier



1235
1236
1237
1238
1239
1240
1241
1242
1243
 
# File 'lib/raptor/http2.rb', line 1235

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_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 (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



1221
1222
1223
1224
1225
 
# File 'lib/raptor/http2.rb', line 1221

def write_http2_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_http2_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 (OpenSSL::SSL::SSLSocket) —

    the connection socket

  • writer (Writer) —

    lock-free 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)

Returns:

  • (String) —

    the response body size in bytes



1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
 
# File 'lib/raptor/http2.rb', line 1167

def write_http2_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