module Worker

parameter　application　Interface(:call)

The HTTP application.

parameter　streams　Array(IO)

The control and bidirectional body streams.

def self.run(application, streams)
	endpoint = Transport::Endpoint.new(*streams)
	
	Sync do
		execute(application, endpoint)
	end
ensure
	endpoint&.close
end

parameter　application　Interface(:call)

The HTTP application.

parameter　endpoint　Transport::Endpoint

The worker endpoint.

def self.execute(application, endpoint)
	response_started = false
	message = endpoint.control.read
	raise ClosedError unless message
	
	type, description = message
	raise ClosedError, "Expected a request, but received #{type.inspect}!" unless type == :request
	
	request = Request.new(description, endpoint)
	response = application.call(request)
	response ||= ::Protocol::HTTP::Response[500]
	
	body = response.body
	body_metadata = body_metadata(body)
	trailers = nil
	
	unless body_metadata
		trailers = response.headers.trailer.to_a
		trailers = nil if trailers.empty?
	end
	
	endpoint.control.write(:response, {
		version: response.version,
		status: response.status,
		headers: response.headers.header.to_a,
		trailers: trailers,
		body: body_metadata,
		protocol: response.protocol,
	})
	response_started = true
	
	if body_metadata
		consume(response, request, endpoint)
	else
		response.close
	end
rescue => error
	if response_started
		write_error(endpoint.body, error)
	else
		write_error(endpoint.control, error)
	end
	
	begin
		response&.close(error)
	rescue
		# The original error has already been reported:
	end
ensure
	request&.close
end

parameter　response　Protocol::HTTP::Response

The response to consume.

parameter　request　Request

The reconstructed request.

parameter　endpoint　Transport::Endpoint

The worker endpoint.

def self.consume(response, request, endpoint)
	message = endpoint.control.read
	raise ClosedError unless message
	
	type, mode = message
	return if type == :cancel
	raise ClosedError, "Expected a consumption mode, but received #{type.inspect}!" unless type == :consume
	
	case mode
	when :read
		response.body.each{|chunk| endpoint.body.write(:chunk, chunk)}
	when :stream
		input = Body::StreamInput.new(endpoint.body, request.headers)
		output = Body::StreamOutput.new(endpoint.body)
		stream = ::Protocol::HTTP::Body::Stream.new(input, output)
		response.body.call(stream)
	else
		raise ArgumentError, "Unknown response consumption mode: #{mode.inspect}!"
	end
	
	trailers = response.headers.trailer.to_a
	endpoint.body.write(:trailers, trailers) unless trailers.empty?
	endpoint.body.close_write
rescue => error
	write_error(endpoint.body, error)
ensure
	response.close(error)
end

parameter　body　Protocol::HTTP::Body::Readable | Nil

The body.

returns　Hash | Nil

The body metadata, or nil for no body.

def self.body_metadata(body)
	return unless body
	return if body.empty?
	
	return {
		length: body.length,
		stream: body.stream?,
	}
end

parameter　channel　Channel

The destination channel.

parameter　error　Exception

The error to report.

def self.write_error(channel, error)
	channel.write(:error, RemoteError.dump(error))
rescue ClosedError
	# The caller has already abandoned the execution:
end