class Async::HTTP::Body::Pipe

def close
	@reader&.stop
	@writer&.stop
	
	@tail.close
end

def close_head
	@head.close
	
	# Both tasks are done, don't keep references:
	@reader = nil
	@writer = nil
end

def initialize(input, output = Writable.new, task: Task.current)
	@input = input
	@output = output
	
	head, tail = ::Socket.pair(Socket::AF_UNIX, Socket::SOCK_STREAM)
	
	@head = ::IO::Stream::Buffered.new(head)
	@tail = tail
	
	@reader = nil
	@writer = nil
	
	task.async(transient: true, &self.method(:reader))
	task.async(transient: true, &self.method(:writer))
end

def reader(task)
	@reader = task
	
	task.annotate "#{self.class} reader."
	
	while chunk = @input.read
		@head.write(chunk)
		@head.flush
	end
	
	@head.close_write
ensure
	@input.close($!)
	
	close_head if @writer&.finished?
end

def to_io
	@tail
end

def writer(task)
	@writer = task
	
	task.annotate "#{self.class} writer."
	
	while chunk = @head.read_partial
		@output.write(chunk)
	end
ensure
	@output.close($!)
	
	close_head if @reader&.finished?
end