class HTTP2::Connection
Connection
encapsulates all of the connection, stream, flow-control, error management, and other processing logic required for a well-behaved HTTP 2.0 endpoint.
Note that this class should not be used directly. Instead, you want to use either Client
or Server
class to drive the HTTP 2.0 exchange.
rubocop:disable ClassLength
Attributes
Number of active streams between client and server (reserved streams are not counted towards the stream limit).
Current settings value for local and peer
Size of current connection flow control window (by default, set to infinity, but is automatically updated on receipt of peer settings).
Pending settings value
Sent but not ack'ed settings
Connection
state (:new, :closed).
Size of current connection flow control window (by default, set to infinity, but is automatically updated on receipt of peer settings).
Public Class Methods
Initializes new connection object.
# File lib/http/2/connection.rb, line 75 def initialize(**settings) @local_settings = DEFAULT_CONNECTION_SETTINGS.merge(settings) @remote_settings = SPEC_DEFAULT_CONNECTION_SETTINGS.dup @compressor = Header::Compressor.new(**settings) @decompressor = Header::Decompressor.new(**settings) @active_stream_count = 0 @streams = {} @streams_recently_closed = {} @pending_settings = [] @framer = Framer.new @local_window_limit = @local_settings[:settings_initial_window_size] @local_window = @local_window_limit @remote_window_limit = @remote_settings[:settings_initial_window_size] @remote_window = @remote_window_limit @recv_buffer = Buffer.new @send_buffer = [] @continuation = [] @error = nil @h2c_upgrade = nil @closed_since = nil end
Public Instance Methods
# File lib/http/2/connection.rb, line 367 def <<(*args) receive(*args) end
# File lib/http/2/connection.rb, line 103 def closed? @state == :closed end
Sends a GOAWAY frame indicating that the peer should stop creating new streams for current connection.
Endpoints MAY append opaque data to the payload of any GOAWAY frame. Additional debug data is intended for diagnostic purposes only and carries no semantic value. Debug data MUST NOT be persistently stored, since it could contain sensitive information.
@param error [Symbol] @param payload [String]
# File lib/http/2/connection.rb, line 141 def goaway(error = :no_error, payload = nil) last_stream = if (max = @streams.max) max.first else 0 end send(type: :goaway, last_stream: last_stream, error: error, payload: payload) @state = :closed @closed_since = Time.now end
Allocates new stream for current connection.
@param priority [Integer] @param window [Integer] @param parent [Stream]
# File lib/http/2/connection.rb, line 112 def new_stream(**args) fail ConnectionClosed if @state == :closed fail StreamLimitExceeded if @active_stream_count >= @remote_settings[:settings_max_concurrent_streams] stream = activate_stream(id: @stream_id, **args) @stream_id += 2 stream end
Sends PING frame to the peer.
@param payload [String] optional payload must be 8 bytes long @param blk [Proc] callback to execute when PONG is received
# File lib/http/2/connection.rb, line 126 def ping(payload, &blk) send(type: :ping, stream: 0, payload: payload) once(:ack, &blk) if blk end
Decodes incoming bytes into HTTP 2.0 frames and routes them to appropriate receivers: connection frames are handled directly, and stream frames are passed to appropriate stream objects.
@param data [String] Binary encoded string
# File lib/http/2/connection.rb, line 179 def receive(data) @recv_buffer << data # Upon establishment of a TCP connection and determination that # HTTP/2.0 will be used by both peers, each endpoint MUST send a # connection header as a final confirmation and to establish the # initial settings for the HTTP/2.0 connection. # # Client connection header is 24 byte connection header followed by # SETTINGS frame. Server connection header is SETTINGS frame only. if @state == :waiting_magic if @recv_buffer.size < 24 if !CONNECTION_PREFACE_MAGIC.start_with? @recv_buffer fail HandshakeError else return # maybe next time end elsif @recv_buffer.read(24) == CONNECTION_PREFACE_MAGIC # MAGIC is OK. Send our settings @state = :waiting_connection_preface payload = @local_settings.reject { |k, v| v == SPEC_DEFAULT_CONNECTION_SETTINGS[k] } settings(payload) else fail HandshakeError end end while (frame = @framer.parse(@recv_buffer)) emit(:frame_received, frame) # Header blocks MUST be transmitted as a contiguous sequence of frames # with no interleaved frames of any other type, or from any other stream. unless @continuation.empty? unless frame[:type] == :continuation && frame[:stream] == @continuation.first[:stream] connection_error end @continuation << frame return unless frame[:flags].include? :end_headers payload = @continuation.map { |f| f[:payload] }.join frame = @continuation.shift @continuation.clear frame.delete(:length) frame[:payload] = Buffer.new(payload) frame[:flags] << :end_headers end # SETTINGS frames always apply to a connection, never a single stream. # The stream identifier for a settings frame MUST be zero. If an # endpoint receives a SETTINGS frame whose stream identifier field is # anything other than 0x0, the endpoint MUST respond with a connection # error (Section 5.4.1) of type PROTOCOL_ERROR. if connection_frame?(frame) connection_management(frame) else case frame[:type] when :headers # When server receives even-numbered stream identifier, # the endpoint MUST respond with a connection error of type PROTOCOL_ERROR. connection_error if frame[:stream].even? && self.is_a?(Server) # The last frame in a sequence of HEADERS/CONTINUATION # frames MUST have the END_HEADERS flag set. unless frame[:flags].include? :end_headers @continuation << frame return end # After sending a GOAWAY frame, the sender can discard frames # for new streams. However, any frames that alter connection # state cannot be completely ignored. For instance, HEADERS, # PUSH_PROMISE and CONTINUATION frames MUST be minimally # processed to ensure a consistent compression state decode_headers(frame) return if @state == :closed stream = @streams[frame[:stream]] if stream.nil? stream = activate_stream( id: frame[:stream], weight: frame[:weight] || DEFAULT_WEIGHT, dependency: frame[:dependency] || 0, exclusive: frame[:exclusive] || false, ) emit(:stream, stream) end stream << frame when :push_promise # The last frame in a sequence of PUSH_PROMISE/CONTINUATION # frames MUST have the END_HEADERS flag set unless frame[:flags].include? :end_headers @continuation << frame return end decode_headers(frame) return if @state == :closed # PUSH_PROMISE frames MUST be associated with an existing, peer- # initiated stream... A receiver MUST treat the receipt of a # PUSH_PROMISE on a stream that is neither "open" nor # "half-closed (local)" as a connection error (Section 5.4.1) of # type PROTOCOL_ERROR. Similarly, a receiver MUST treat the # receipt of a PUSH_PROMISE that promises an illegal stream # identifier (Section 5.1.1) (that is, an identifier for a stream # that is not currently in the "idle" state) as a connection error # (Section 5.4.1) of type PROTOCOL_ERROR, unless the receiver # recently sent a RST_STREAM frame to cancel the associated stream. parent = @streams[frame[:stream]] pid = frame[:promise_stream] # if PUSH parent is recently closed, RST_STREAM the push if @streams_recently_closed[frame[:stream]] send(type: :rst_stream, stream: pid, error: :refused_stream) return end connection_error(msg: 'missing parent ID') if parent.nil? unless parent.state == :open || parent.state == :half_closed_local # An endpoint might receive a PUSH_PROMISE frame after it sends # RST_STREAM. PUSH_PROMISE causes a stream to become "reserved". # The RST_STREAM does not cancel any promised stream. Therefore, if # promised streams are not desired, a RST_STREAM can be used to # close any of those streams. if parent.closed == :local_rst # We can either (a) 'resurrect' the parent, or (b) RST_STREAM # ... sticking with (b), might need to revisit later. send(type: :rst_stream, stream: pid, error: :refused_stream) else connection_error end end stream = activate_stream(id: pid, parent: parent) emit(:promise, stream) stream << frame else if (stream = @streams[frame[:stream]]) stream << frame if frame[:type] == :data update_local_window(frame) calculate_window_update(@local_window_limit) end else case frame[:type] # The PRIORITY frame can be sent for a stream in the "idle" or # "closed" state. This allows for the reprioritization of a # group of dependent streams by altering the priority of an # unused or closed parent stream. when :priority stream = activate_stream( id: frame[:stream], weight: frame[:weight] || DEFAULT_WEIGHT, dependency: frame[:dependency] || 0, exclusive: frame[:exclusive] || false, ) emit(:stream, stream) stream << frame # WINDOW_UPDATE can be sent by a peer that has sent a frame # bearing the END_STREAM flag. This means that a receiver could # receive a WINDOW_UPDATE frame on a "half-closed (remote)" or # "closed" stream. A receiver MUST NOT treat this as an error # (see Section 5.1). when :window_update process_window_update(frame) else # An endpoint that receives an unexpected stream identifier # MUST respond with a connection error of type PROTOCOL_ERROR. connection_error end end end end end rescue StandardError => e raise if e.is_a?(Error::Error) connection_error(e: e) end
Sends a connection SETTINGS frame to the peer. The values are reflected when the corresponding ACK is received.
@param settings [Array or Hash]
# File lib/http/2/connection.rb, line 166 def settings(payload) payload = payload.to_a connection_error if validate_settings(@local_role, payload) @pending_settings << payload send(type: :settings, stream: 0, payload: payload) @pending_settings << payload end
Sends a WINDOW_UPDATE frame to the peer.
@param increment [Integer]
# File lib/http/2/connection.rb, line 157 def window_update(increment) @local_window += increment send(type: :window_update, stream: 0, increment: increment) end
Private Instance Methods
Activates new incoming or outgoing stream and registers appropriate connection managemet callbacks.
@param id [Integer] @param priority [Integer] @param window [Integer] @param parent [Stream]
# File lib/http/2/connection.rb, line 675 def activate_stream(id: nil, **args) connection_error(msg: 'Stream ID already exists') if @streams.key?(id) stream = Stream.new(**{ connection: self, id: id }.merge(args)) # Streams that are in the "open" state, or either of the "half closed" # states count toward the maximum number of streams that an endpoint is # permitted to open. stream.once(:active) { @active_stream_count += 1 } stream.once(:close) do @active_stream_count -= 1 # Store a reference to the closed stream, such that we can respond # to any in-flight frames while close is registered on both sides. # References to such streams will be purged whenever another stream # is closed, with a defined RTT time window. @streams_recently_closed[id] = Time.now.to_i cleanup_recently_closed end stream.on(:promise, &method(:promise)) if self.is_a? Server stream.on(:frame, &method(:send)) @streams[id] = stream end
Purge recently streams closed within defined RTT time window.
# File lib/http/2/connection.rb, line 702 def cleanup_recently_closed now_ts = Time.now.to_i to_delete = [] @streams_recently_closed.each do |stream_id, ts| # Ruby Hash enumeration is ordered, so once fresh stream is met we can stop searching. break if now_ts - ts < RECENTLY_CLOSED_STREAMS_TTL to_delete << stream_id end to_delete.each do |stream_id| @streams.delete stream_id @streams_recently_closed.delete stream_id end end
Emit GOAWAY error indicating to peer that the connection is being aborted, and once sent, raise a local exception.
@param error [Symbol] @option error [Symbol] :no_error @option error [Symbol] :internal_error @option error [Symbol] :flow_control_error @option error [Symbol] :stream_closed @option error [Symbol] :frame_too_large @option error [Symbol] :compression_error @param msg [String]
# File lib/http/2/connection.rb, line 728 def connection_error(error = :protocol_error, msg: nil, e: nil) goaway(error) unless @state == :closed || @state == :new @state, @error = :closed, error klass = error.to_s.split('_').map(&:capitalize).join msg ||= e && e.message backtrace = (e && e.backtrace) || [] fail Error.const_get(klass), msg, backtrace end
Check if frame is a connection frame: SETTINGS, PING, GOAWAY, and any frame addressed to stream ID = 0.
@param frame [Hash] @return [Boolean]
# File lib/http/2/connection.rb, line 417 def connection_frame?(frame) (frame[:stream]).zero? || frame[:type] == :settings || frame[:type] == :ping || frame[:type] == :goaway end
Process received connection frame (stream ID = 0).
-
Handle SETTINGS updates
-
Connection
flow control (WINDOW_UPDATE) -
Emit PONG auto-reply to PING frames
-
Mark connection as closed on GOAWAY
@param frame [Hash]
# File lib/http/2/connection.rb, line 431 def connection_management(frame) case @state when :waiting_connection_preface # The first frame MUST be a SETTINGS frame at the start of a connection. @state = :connected connection_settings(frame) when :connected case frame[:type] when :settings connection_settings(frame) when :window_update @remote_window += frame[:increment] send_data(nil, true) when :ping if frame[:flags].include? :ack emit(:ack, frame[:payload]) else send(type: :ping, stream: 0, flags: [:ack], payload: frame[:payload]) end when :goaway # Receivers of a GOAWAY frame MUST NOT open additional streams on # the connection, although a new connection can be established # for new streams. @state = :closed @closed_since = Time.now emit(:goaway, frame[:last_stream], frame[:error], frame[:payload]) when :altsvc # 4. The ALTSVC HTTP/2 Frame # An ALTSVC frame on stream 0 with empty (length 0) "Origin" # information is invalid and MUST be ignored. if frame[:origin] && !frame[:origin].empty? emit(frame[:type], frame) end when :blocked emit(frame[:type], frame) else connection_error end when :closed connection_error if (Time.now - @closed_since) > 15 else connection_error end end
Update connection settings based on parameters set by the peer.
@param frame [Hash]
# File lib/http/2/connection.rb, line 531 def connection_settings(frame) connection_error unless frame[:type] == :settings && (frame[:stream]).zero? # Apply settings. # side = # local: previously sent and pended our settings should be effective # remote: just received peer settings should immediately be effective settings, side = if frame[:flags].include?(:ack) # Process pending settings we have sent. [@pending_settings.shift, :local] else connection_error if validate_settings(@remote_role, frame[:payload]) [frame[:payload], :remote] end settings.each do |key, v| case side when :local @local_settings[key] = v when :remote @remote_settings[key] = v end case key when :settings_max_concurrent_streams # Do nothing. # The value controls at the next attempt of stream creation. when :settings_initial_window_size # A change to SETTINGS_INITIAL_WINDOW_SIZE could cause the available # space in a flow control window to become negative. A sender MUST # track the negative flow control window, and MUST NOT send new flow # controlled frames until it receives WINDOW_UPDATE frames that cause # the flow control window to become positive. case side when :local @local_window = @local_window - @local_window_limit + v @streams.each do |_id, stream| stream.emit(:local_window, stream.local_window - @local_window_limit + v) end @local_window_limit = v when :remote @remote_window = @remote_window - @remote_window_limit + v @streams.each do |_id, stream| # Event name is :window, not :remote_window stream.emit(:window, stream.remote_window - @remote_window_limit + v) end @remote_window_limit = v end when :settings_header_table_size # Setting header table size might cause some headers evicted case side when :local @compressor.table_size = v when :remote @decompressor.table_size = v end when :settings_enable_push # nothing to do when :settings_max_frame_size # update framer max_frame_size @framer.max_frame_size = v # else # ignore unknown settings end end case side when :local # Received a settings_ack. Notify application layer. emit(:settings_ack, frame, @pending_settings.size) when :remote unless @state == :closed || @h2c_upgrade == :start # Send ack to peer send(type: :settings, stream: 0, payload: [], flags: [:ack]) end end end
Decode headers payload and update connection decompressor state.
The receiver endpoint reassembles the header block by concatenating the individual fragments, then decompresses the block to reconstruct the header set - aka, header payloads are buffered until END_HEADERS, or an END_PROMISE flag is seen.
@param frame [Hash]
# File lib/http/2/connection.rb, line 623 def decode_headers(frame) if frame[:payload].is_a? Buffer frame[:payload] = @decompressor.decode(frame[:payload]) end rescue CompressionError => e connection_error(:compression_error, e: e) rescue ProtocolError => e connection_error(:protocol_error, e: e) rescue StandardError => e connection_error(:internal_error, e: e) end
Applies HTTP 2.0 binary encoding to the frame.
@param frame [Hash] @return [Array of Buffer] encoded frame
# File lib/http/2/connection.rb, line 402 def encode(frame) frames = if frame[:type] == :headers || frame[:type] == :push_promise encode_headers(frame) # HEADERS and PUSH_PROMISE may create more than one frame else [frame] # otherwise one frame end frames.map { |f| @framer.generate(f) } end
Encode headers payload and update connection compressor state.
@param frame [Hash] @return [Array of Frame]
# File lib/http/2/connection.rb, line 640 def encode_headers(frame) payload = frame[:payload] payload = @compressor.encode(payload) unless payload.is_a? Buffer frames = [] while payload.bytesize > 0 cont = frame.dup cont[:type] = :continuation cont[:flags] = [] cont[:payload] = payload.slice!(0, @remote_settings[:settings_max_frame_size]) frames << cont end if frames.empty? frames = [frame] else frames.first[:type] = frame[:type] frames.first[:flags] = frame[:flags] - [:end_headers] frames.last[:flags] << :end_headers end frames rescue StandardError => e connection_error(:compression_error, e: e) nil end
# File lib/http/2/connection.rb, line 739 def manage_state(_) yield end
Send an outgoing frame. DATA frames are subject to connection flow control and may be split and / or buffered based on current window size. All other frames are sent immediately.
@note all frames are currently delivered in FIFO order. @param frame [Hash]
# File lib/http/2/connection.rb, line 379 def send(frame) emit(:frame_sent, frame) if frame[:type] == :data send_data(frame, true) else # An endpoint can end a connection at any time. In particular, an # endpoint MAY choose to treat a stream error as a connection error. if frame[:type] == :rst_stream && frame[:error] == :protocol_error goaway(frame[:error]) else # HEADERS and PUSH_PROMISE may generate CONTINUATION. Also send # RST_STREAM that are not protocol errors frames = encode(frame) frames.each { |f| emit(:frame, f) } end end end
Validate settings parameters. See sepc Section 6.5.2.
@param role [Symbol] The sender's role: :client or :server @return nil if no error. Exception object in case of any error.
# File lib/http/2/connection.rb, line 482 def validate_settings(role, settings) settings.each do |key, v| case key when :settings_header_table_size # Any value is valid when :settings_enable_push case role when :server # Section 8.2 # Clients MUST reject any attempt to change the # SETTINGS_ENABLE_PUSH setting to a value other than 0 by treating the # message as a connection error (Section 5.4.1) of type PROTOCOL_ERROR. return ProtocolError.new("invalid #{key} value") unless v.zero? when :client # Any value other than 0 or 1 MUST be treated as a # connection error (Section 5.4.1) of type PROTOCOL_ERROR. unless v.zero? || v == 1 return ProtocolError.new("invalid #{key} value") end end when :settings_max_concurrent_streams # Any value is valid when :settings_initial_window_size # Values above the maximum flow control window size of 2^31-1 MUST # be treated as a connection error (Section 5.4.1) of type # FLOW_CONTROL_ERROR. unless v <= 0x7fffffff return FlowControlError.new("invalid #{key} value") end when :settings_max_frame_size # The initial value is 2^14 (16,384) octets. The value advertised # by an endpoint MUST be between this initial value and the maximum # allowed frame size (2^24-1 or 16,777,215 octets), inclusive. # Values outside this range MUST be treated as a connection error # (Section 5.4.1) of type PROTOCOL_ERROR. unless v >= 16_384 && v <= 16_777_215 return ProtocolError.new("invalid #{key} value") end when :settings_max_header_list_size # Any value is valid # else # ignore unknown settings end end nil end