class Async::HTTP::Protocol::HTTP2

def self.client(stream)
	self.new(::HTTP2::Client.new, stream)
end

def self.server(stream)
	self.new(::HTTP2::Server.new, stream)
end

def call(request)
	request.version ||= self.version
	
	stream = @controller.new_stream
	@count += 1
	
	headers = {
		SCHEME => HTTPS,
		METHOD => request.method.to_s,
		PATH => request.path.to_s,
		AUTHORITY => request.authority.to_s,
	}.merge(request.headers)
	
	finished = Async::Notification.new
	
	exception = nil
	response = Response.new
	response.version = self.version
	response.headers = {}
	body = Body::Writable.new
	response.body = body
	
	stream.on(:headers) do |headers|
		headers.each do |key, value|
			if key == STATUS
				response.status = value.to_i
			elsif key == REASON
				response.reason = value
			else
				response.headers[key] = value
			end
		end
		
		# At this point, we are now expecting two events: data and close.
		stream.on(:close) do |error|
			# If we receive close after this point, it's not a request error, but a failure we need to signal to the body.
			if error
				body.stop(EOFError.new(error))
			else
				body.finish
			end
		end
		
		finished.signal
	end
	
	stream.on(:data) do |chunk|
		body.write(chunk.to_s) unless chunk.empty?
	end
	
	stream.on(:close) do |error|
		# The remote server has closed the connection while we were sending the request.
		if error
			exception = EOFError.new(error)
			finished.signal
		end
	end
	
	if request.body.nil? or request.body.empty?
		stream.headers(headers, end_stream: true)
		request.body.read if request.body
	else
		begin
			stream.headers(headers, end_stream: false)
		rescue
			raise RequestFailed.new
		end
		
		request.body.each do |chunk|
			stream.data(chunk, end_stream: false)
		end
			
		stream.data("", end_stream: true)
	end
	
	start_connection
	@stream.flush
	
	Async.logger.debug(self) {"Stream flushed, waiting for signal."}
	finished.wait
	
	if exception
		raise exception
	end
	
	Async.logger.debug(self) {"Stream finished: #{response.inspect}"}
	return response
end

def close
	Async.logger.debug(self) {"Closing connection"}
	
	@reader.stop if @reader
	@stream.close
end

def good?
	@stream.connected?
end

def initialize(controller, stream)
	@controller = controller
	@stream = stream
	
	@controller.on(:frame) do |data|
		@stream.write(data)
		@stream.flush
	end
	
	@controller.on(:frame_sent) do |frame|
		Async.logger.debug(self) {"Sent frame: #{frame.inspect}"}
	end
	
	@controller.on(:frame_received) do |frame|
		Async.logger.debug(self) {"Received frame: #{frame.inspect}"}
	end
	
	@controller.on(:goaway) do |payload|
		Async.logger.error(self) {"goaway: #{payload.inspect}"}
		
		@reader.stop
		@stream.close
	end
	
	@count = 0
end

def multiplex
	@controller.remote_settings[:settings_max_concurrent_streams]
end

def read_in_background(task: Task.current)
	task.async do |nested_task|
		nested_task.annotate("#{version} reading data")
		
		while buffer = @stream.read_partial
			@controller << buffer
		end
		
		Async.logger.debug(self) {"Connection reset by peer!"}
	end
end

def receive_requests(task: Task.current, &block)
	# emits new streams opened by the client
	@controller.on(:stream) do |stream|
		request = Request.new
		request.version = self.version
		request.headers = Headers.new
		body = Body::Writable.new
		request.body = body
		
		stream.on(:headers) do |headers|
			headers.each do |key, value|
				if key == METHOD
					request.method = value
				elsif key == PATH
					request.path = value
				elsif key == AUTHORITY
					request.authority = value
				else
					request.headers[key] = value
				end
			end
		end
		
		stream.on(:data) do |chunk|
			# puts "Got request data: #{chunk.inspect}"
			body.write(chunk.to_s) unless chunk.empty?
		end
		
		stream.on(:close) do |error|
			if error
				body.stop(EOFError.new(error))
			end
		end
		
		stream.on(:half_close) do
			# The requirements for this to be in lock-step with other opertaions is minimal.
			# TODO consider putting this in it's own async task.
			begin
				# We are no longer receiving any more data frames:
				body.finish
				
				# Generate the response:
				response = yield request
				
				headers = {STATUS => response.status.to_s}
				headers.update(response.headers)
				
				if response.body.nil? or response.body.empty?
					stream.headers(headers, end_stream: true)
					response.body.read if response.body
				else
					stream.headers(headers, end_stream: false)
					
					response.body.each do |chunk|
						stream.data(chunk, end_stream: false)
					end
					
					stream.data("", end_stream: true)
				end
			rescue
				Async.logger.error(self) {$!}
				
				# Generating the response failed.
				stream.close(:internal_error)
			end
		end
	end
	
	start_connection
	@reader.wait
end

def reusable?
	!@stream.closed?
end

def start_connection
	@reader ||= read_in_background
end

def version
	VERSION
end