module Connection

Provides shared connection behaviour for HTTP/2 client and server connections.

Definitions

def initialize(...)

Initialize the connection state.

Implementation

 
def initialize(...)
	super
	
	@reader = nil
	
	# Writing multiple frames at the same time can cause odd problems if frames are only partially written. So we use a semaphore to ensure frames are written in their entirety.
	@write_frame_guard = Async::Semaphore.new(1)
end

def synchronize(&block)

Synchronize write access to the connection.

Signature
yields　{|...| ...}

The block to execute while holding the write lock.

Implementation

 
def synchronize(&block)
	@write_frame_guard.acquire(&block)
end

def write_frame(frame)

Write a single frame, deferring stop until the frame is written.

Implementation

 
def write_frame(frame)
	Task.current.defer_stop{super}
end

def write_frames

Write multiple frames, deferring stop until the frames are written.

Implementation

 
def write_frames
	Task.current.defer_stop{super}
end

def to_s

Signature
returns　String

A string representation of this connection.

Implementation

 
def to_s
	"\#<#{self.class} #{@streams.count} active streams>"
end

def as_json(...)

Signature
returns　String

A JSON-compatible representation.

Implementation

 
def as_json(...)
	to_s
end

def to_json(...)

Signature
returns　String

A JSON string representation.

Implementation

 
def to_json(...)
	as_json.to_json(...)
end

def http1?

Signature
returns　Boolean

Whether this is an HTTP/1 connection.

Implementation

 
def http1?
	false
end

def http2?

Signature
returns　Boolean

Whether this is an HTTP/2 connection.

Implementation

 
def http2?
	true
end

def start_connection

Start the background reader task if it is not already running.

Implementation

 
def start_connection
	@reader || read_in_background
end

def close(error = nil)

Close the connection and stop the background reader.

Implementation

 
def close(error = nil)
	if reader = @reader
		@reader = nil
		
		# The reader task can close the connection itself, e.g. when the last stream completes and the connection is released back to the pool. Stopping it here would cancel the current task in the middle of this method, leaving the underlying stream open, so we let it unwind by itself: `closed?` is now true, so the read loop exits.
		reader.stop unless reader.current?
	end
	
	super
end

def close_if_drained!

The connection has finished draining the streams which the remote peer accepted before its graceful GOAWAY.

The last of those streams can complete on the sending side, in a task other than the background reader - the response arrived first and the request body was still being written. The reader is then parked in a blocking read and will never notice that the connection is closed, so we stop it and let its ensure close the connection.

Implementation

 
def close_if_drained!
	super
	
	if self.closed? and (reader = @reader) and !reader.current?
		reader.stop
	end
end

def read_in_background(parent: Task.current)

Start a transient background task that reads frames from the connection.

Implementation

 
def read_in_background(parent: Task.current)
	raise RuntimeError, "Connection is closed!" if closed?
	
	parent.async(transient: true) do |task|
		@reader = task
		
		task.annotate("#{version} reading data for #{self.class}.")
		
		# We don't need to defer stop here as this is already a transient task (ignores stop):
		begin
			while !self.closed?
				self.consume_window
				self.read_frame
			end
		rescue => error
			# Close with error.
		ensure
			# Don't call #close twice.
			if @reader
				@reader = nil
				
				self.close(error)
			end
		end
	end
end

def peer

Signature
returns　Protocol::HTTP::Peer

The peer information for this connection.

Implementation

 
def peer
	@peer ||= ::Protocol::HTTP::Peer.for(@stream.io)
end

def concurrency

Signature
returns　Integer

The maximum number of concurrent streams allowed.

Implementation

 
def concurrency
	self.maximum_concurrent_streams
end

def viable?

Can we use this connection to make requests?

Implementation

 
def viable?
	!self.goaway_received? && @stream&.readable?
end

def reusable?

Once the remote peer has sent a GOAWAY frame, it will not process any new streams on this connection, so it must not be handed out for another request, even while the streams it accepted are still being drained.

Signature
returns　Boolean

Whether the connection can be reused.

Implementation

 
def reusable?
	!self.closed? && !self.goaway_received?
end

def version

Signature
returns　String

The HTTP version string.

Implementation

 
def version
	VERSION
end