class Connection

parameter　framer　Framer

The frame handler for reading/writing HTTP/2 frames.

parameter　local_stream_id　Integer

The starting stream ID for locally-initiated streams.

def initialize(framer, local_stream_id)
	super()
	
	@state = :new
	
	# Hash(Integer, Stream)
	@streams = {}
	
	@framer = framer
	
	# The next stream id to use:
	@local_stream_id = local_stream_id
	
	# The biggest remote stream id seen thus far:
	@remote_stream_id = 0
	
	@local_settings = PendingSettings.new
	@remote_settings = Settings.new
	
	@decoder = HPACK::Context.new
	@encoder = HPACK::Context.new
	
	@local_window = LocalWindow.new
	@remote_window = Window.new
end

returns　Integer

Always returns 0 for the connection itself.

def id
	0
end

parameter　id　Integer

The stream ID to look up.

returns　Connection | Stream | Nil

The connection (if id=0), stream, or nil.

def [] id
	if id.zero?
		self
	else
		@streams[id]
	end
end

def maximum_frame_size
	@remote_settings.maximum_frame_size
end

def maximum_concurrent_streams
	@remote_settings.maximum_concurrent_streams
end

def closed?
	@state == :closed || @framer.nil?
end

parameter　id　Integer

The stream ID to remove.

returns　Stream | Nil

The removed stream, or nil if not found.

def delete(id)
	@streams.delete(id)
end

def close(error = nil)
	# The underlying socket may already be closed by this point.
	@streams.each_value{|stream| stream.close(error)}
	@streams.clear
	
ensure
	if @framer
		@framer.close
		@framer = nil
	end
end

parameter　headers　Array

The headers to encode.

parameter　buffer　String

Optional buffer for encoding output.

returns　String

The encoded header block.

def encode_headers(headers, buffer = String.new.b)
	HPACK::Compressor.new(buffer, @encoder, table_size_limit: @remote_settings.header_table_size).encode(headers)
end

parameter　data　String

The encoded header block data.

returns　Array

The decoded headers.

def decode_headers(data)
	HPACK::Decompressor.new(data, @decoder, table_size_limit: @local_settings.header_table_size).decode
end

def next_stream_id
	id = @local_stream_id
	
	@local_stream_id += 2
	
	return id
end

def ignore_frame?(frame)
	if self.closed?
		# puts "ignore_frame? #{frame.stream_id} -> #{valid_remote_stream_id?(frame.stream_id)} > #{@remote_stream_id}"
		if valid_remote_stream_id?(frame.stream_id)
			return frame.stream_id > @remote_stream_id
		end
	end
end

def synchronize
	yield
end

def read_frame
	frame = @framer.read_frame(@local_settings.maximum_frame_size)
	
	# puts "#{self.class} #{@state} read_frame: class=#{frame.class} stream_id=#{frame.stream_id} flags=#{frame.flags} length=#{frame.length} (remote_stream_id=#{@remote_stream_id})"
	# puts "Windows: local_window=#{@local_window.inspect}; remote_window=#{@remote_window.inspect}"
	
	return if ignore_frame?(frame)
	
	yield frame if block_given?
	frame.apply(self)
	
	return frame
rescue GoawayError => error
	# Go directly to jail. Do not pass go, do not collect $200.
	raise
rescue ProtocolError => error
	send_goaway(error.code || PROTOCOL_ERROR, error.message)
	
	raise
rescue HPACK::Error => error
	send_goaway(COMPRESSION_ERROR, error.message)
	
	raise
end

parameter　changes　Hash

The settings changes to send.

def send_settings(changes)
	@local_settings.append(changes)
	
	frame = SettingsFrame.new
	frame.pack(changes)
	
	write_frame(frame)
end

def close!
	@state = :closed
	
	return self
end

def send_goaway(error_code = 0, message = "")
	frame = GoawayFrame.new
	frame.pack @remote_stream_id, error_code, message
	
	write_frame(frame)
ensure
	self.close!
end

parameter　frame　GoawayFrame

The GOAWAY frame to process.

raises　GoawayError

If the frame indicates a connection error.

def receive_goaway(frame)
	# We capture the last stream that was processed.
	@remote_stream_id, error_code, message = frame.unpack
	
	self.close!
	
	if error_code != 0
		# Shut down immediately.
		raise GoawayError.new(message, error_code)
	end
end

parameter　frame　Frame

The frame to write.

def write_frame(frame)
	synchronize do
		@framer.write_frame(frame)
	end
	
	@framer.flush
end

yields　{|framer| ...}

The framer for writing multiple frames.

parameter　framer　Framer

The framer instance.

raises　EOFError

If the connection is closed.

def write_frames
	if @framer
		synchronize do
			yield @framer
		end
		
		@framer.flush
	else
		raise EOFError, "Connection closed!"
	end
end

parameter　changes　Hash

The settings changes to apply locally.

def update_local_settings(changes)
	capacity = @local_settings.initial_window_size
	
	@streams.each_value do |stream|
		stream.local_window.capacity = capacity
	end
	
	@local_window.desired = capacity
end

parameter　changes　Hash

The settings changes to apply to remote peer.

def update_remote_settings(changes)
	capacity = @remote_settings.initial_window_size
	
	@streams.each_value do |stream|
		stream.remote_window.capacity = capacity
	end
end

def process_settings(frame)
	if frame.acknowledgement?
		# The remote end has confirmed the settings have been received:
		changes = @local_settings.acknowledge
		
		update_local_settings(changes)
		
		return true
	else
		# The remote end is updating the settings, we reply with acknowledgement:
		reply = frame.acknowledge
		
		write_frame(reply)
		
		changes = frame.unpack
		@remote_settings.update(changes)
		
		update_remote_settings(changes)
		
		return false
	end
end

returns　Connection

Self for method chaining.

def open!
	@state = :open
	
	return self
end

parameter　frame　SettingsFrame

The settings frame to process.

raises　ProtocolError

If the connection is in an invalid state.

def receive_settings(frame)
	if @state == :new
		# We transition to :open when we receive acknowledgement of first settings frame:
		open! if process_settings(frame)
	elsif @state != :closed
		process_settings(frame)
	else
		raise ProtocolError, "Cannot receive settings in state #{@state}"
	end
end

parameter　data　String

The 8-byte ping payload data.

def send_ping(data)
	if @state != :closed
		frame = PingFrame.new
		frame.pack data
		
		write_frame(frame)
	else
		raise ProtocolError, "Cannot send ping in state #{@state}"
	end
end

parameter　frame　PingFrame

The ping frame to process.

raises　ProtocolError

If ping is received in invalid state.

def receive_ping(frame)
	if @state != :closed
		# This is handled in `read_payload`:
		# if frame.stream_id != 0
		# 	raise ProtocolError, "Ping received for non-zero stream!"
		# end
		
		unless frame.acknowledgement?
			reply = frame.acknowledge
			
			write_frame(reply)
		end
	else
		raise ProtocolError, "Cannot receive ping in state #{@state}"
	end
end

parameter　frame　DataFrame

The data frame to process.

raises　ProtocolError

If data is received for invalid stream.

def receive_data(frame)
	update_local_window(frame)
	
	if stream = @streams[frame.stream_id]
		stream.receive_data(frame)
	elsif closed_stream_id?(frame.stream_id)
		# This can occur if one end sent a stream reset, while the other end was sending a data frame. It's mostly harmless.
	else
		raise ProtocolError, "Cannot receive data for stream id #{frame.stream_id}"
	end
end

parameter　stream_id　Integer

The stream ID to validate.

returns　Boolean

True if the stream ID is valid for remote initiation.

def valid_remote_stream_id?(stream_id)
	false
end

def accept_stream(stream_id, &block)
	unless valid_remote_stream_id?(stream_id)
		raise ProtocolError, "Invalid stream id: #{stream_id}"
	end
	
	create_stream(stream_id, &block)
end

def accept_push_promise_stream(stream_id, &block)
	accept_stream(stream_id, &block)
end

def create_stream(id = next_stream_id, &block)
	if @streams.key?(id)
		raise ProtocolError, "Cannot create stream with id #{id}, already exists!"
	end
	
	if block_given?
		return yield(self, id)
	else
		return Stream.create(self, id)
	end
end

yields　{|stream| ...}

Optional block to configure the created stream.

returns　Stream

The created push promise stream.

def create_push_promise_stream(&block)
	create_stream(&block)
end

def receive_headers(frame)
	stream_id = frame.stream_id
	
	if stream_id.zero?
		raise ProtocolError, "Cannot receive headers for stream 0!"
	end
	
	if stream = @streams[stream_id]
		stream.receive_headers(frame)
	else
		if stream_id <= @remote_stream_id
			raise ProtocolError, "Invalid stream id: #{stream_id} <= #{@remote_stream_id}!"
		end
		
		# We need to validate that we have less streams than the specified maximum:
		if @streams.size < @local_settings.maximum_concurrent_streams
			stream = accept_stream(stream_id)
			@remote_stream_id = stream_id
			
			stream.receive_headers(frame)
		else
			raise ProtocolError, "Exceeded maximum concurrent streams"
		end
	end
end

parameter　frame　PushPromiseFrame

The push promise frame.

raises　ProtocolError

Always raises as push promises are not supported.

def receive_push_promise(frame)
	raise ProtocolError, "Unable to receive push promise!"
end

parameter　frame　PriorityUpdateFrame

The priority update frame.

raises　ProtocolError

If the stream ID is invalid.

def receive_priority_update(frame)
	if frame.stream_id != 0
		raise ProtocolError, "Invalid stream id: #{frame.stream_id}"
	end
	
	stream_id, value = frame.unpack
	
	# Apparently you can set the priority of idle streams, but I'm not sure why that makes sense, so for now let's ignore it.
	if stream = @streams[stream_id]
		stream.priority = Protocol::HTTP::Header::Priority.new(value)
	end
end

parameter　id　Integer

The stream ID to check.

returns　Boolean

True if the stream ID is client-initiated.

def client_stream_id?(id)
	id.odd?
end

parameter　id　Integer

The stream ID to check.

returns　Boolean

True if the stream ID is server-initiated.

def server_stream_id?(id)
	id.even?
end

parameter　id　Integer

The stream ID to check.

returns　Boolean

True if the stream ID is idle (not yet used).

def idle_stream_id?(id)
	if id.even?
		# Server-initiated streams are even.
		if @local_stream_id.even?
			id >= @local_stream_id
		else
			id > @remote_stream_id
		end
	elsif id.odd?
		# Client-initiated streams are odd.
		if @local_stream_id.odd?
			id >= @local_stream_id
		else
			id > @remote_stream_id
		end
	end
end

def closed_stream_id?(id)
	if id.zero?
		# The connection "stream id" can never be closed:
		false
	else
		!idle_stream_id?(id)
	end
end

parameter　frame　ResetStreamFrame

The reset stream frame.

raises　ProtocolError

If the frame is invalid for connection context.

def receive_reset_stream(frame)
	if frame.connection?
		raise ProtocolError, "Cannot reset connection!"
	elsif stream = @streams[frame.stream_id]
		stream.receive_reset_stream(frame)
	elsif closed_stream_id?(frame.stream_id)
		# Ignore.
	else
		raise StreamClosed, "Cannot reset stream #{frame.stream_id}"
	end
end

parameter　amount　Integer

the amount of data to write. Defaults to the current window capacity.

def consume_window(size = self.available_size)
	# Return if there is no window to consume:
	return unless size > 0
	
	@streams.each_value do |stream|
		if stream.active?
			stream.window_updated(size)
		end
	end
end

parameter　frame　WindowUpdateFrame

The window update frame.

def receive_window_update(frame)
	if frame.connection?
		super
		
		self.consume_window
	elsif stream = @streams[frame.stream_id]
		begin
			stream.receive_window_update(frame)
		rescue ProtocolError => error
			stream.send_reset_stream(error.code)
		end
	elsif closed_stream_id?(frame.stream_id)
		# Ignore.
	else
		# Receiving any frame other than HEADERS or PRIORITY on a stream in this state (idle) MUST be treated as a connection error of type PROTOCOL_ERROR.
		raise ProtocolError, "Cannot update window of idle stream #{frame.stream_id}"
	end
end

parameter　frame　ContinuationFrame

The continuation frame.

raises　ProtocolError

Always raises as unexpected continuation frames are not supported.

def receive_continuation(frame)
	raise ProtocolError, "Received unexpected continuation: #{frame.class}"
end

parameter　frame　Frame

The frame to receive.

def receive_frame(frame)
	# ignore.
end