Merge 7c14963 into f79be6f

async-io.gemspec

@@ -17,7 +17,7 @@ Gem::Specification.new do |spec|
 	spec.has_rdoc      = "yard"
 
 	spec.add_dependency "async", "~> 1.3"
-	spec.add_development_dependency "async-rspec", "~> 1.6"
+	spec.add_development_dependency "async-rspec", "~> 1.7"
 
 	spec.required_ruby_version = '~> 2.3'
 

lib/async/io/stream.rb

@@ -38,6 +38,7 @@ def initialize(io, block_size: BLOCK_SIZE, sync: true)
 				@block_size = block_size
 
 				@read_buffer = BinaryString.new
+				@output_buffer = BinaryString.new
 				@write_buffer = BinaryString.new
 			end
 
@@ -180,19 +181,14 @@ def eof!
 
 			# Fills the buffer from the underlying stream.
 			def fill_read_buffer
-				# Can we read directly into the buffer? (Ruby doesn't support append, only replace):
-				if @read_buffer.empty?
-					if @io.read(@block_size, @read_buffer)
-						return true
-					end
-				elsif chunk = @io.read(@block_size)
+				if chunk = @io.read(@block_size, @output_buffer)
 					@read_buffer << chunk
 					return true
+				else
+					# We didn't read anything, so we must be at eof:
+					@eof = true
+					return false
 				end
-
-				# We didn't read anything, so we must be at eof:
-				@eof = true
-				return false
 			end
 
 			# Consumes at most `size` bytes from the buffer.
@@ -205,8 +201,8 @@ def consume_read_buffer(size = nil)
 
 				if size == nil || size >= @read_buffer.size
 					# Consume the entire read buffer:
-					result = @read_buffer.dup
-					@read_buffer.clear
+					result = @read_buffer
+					@read_buffer = BinaryString.new
 				else
 					# Consume only part of the read buffer:
 					result = @read_buffer.slice!(0, size)

spec/async/io/stream_spec.rb

@@ -21,6 +21,8 @@
 require 'async/io/stream'
 
 RSpec.describe Async::IO::Stream do
+	include_context Async::RSpec::Memory
+
 	let(:io) {StringIO.new}
 	let(:stream) {Async::IO::Stream.new(io)}
 
@@ -47,11 +49,19 @@
 			expect(stream.read(20)).to be == "o World"
 			expect(stream).to be_eof
 		end
+
+		context "with large content" do
+			let!(:io) { StringIO.new("a" * 5*1024*1024) }
+
+			it "allocates expected amount of bytes" do
+				expect do
+					stream.read(16*1024).clear until stream.eof?
+				end.to limit_allocations(size: 100*1024)
+			end
+		end
 	end
 
 	describe '#read_until' do
-		include_context Async::RSpec::Memory
-
 		it "can read a line" do
 			io.write("hello\nworld\n")
 			io.seek(0)
@@ -61,13 +71,14 @@
 			expect(stream.read_until("\n")).to be_nil
 		end
 
-		it "minimises allocations" do
-			io.write("hello\nworld\n")
-			io.seek(0)
+		context "with large content" do
+			let!(:io) { StringIO.new("a" * 5*1024*1024 + "b") }
 
-			expect do
-				stream.read_until("\n")
-			end.to limit_allocations(String => 3)
+			it "allocates expected amount of bytes" do
+				expect do
+					stream.read_until("b").clear
+				end.to limit_allocations(size: 100*1024)
+			end
 		end
 	end
 
@@ -96,6 +107,16 @@
 
 			expect(stream.read_partial(11)).to be == "Hello World"
 		end
+
+		context "with large content" do
+			let!(:io) { StringIO.new("a" * 5*1024*1024) }
+
+			it "allocates expected amount of bytes" do
+				expect do
+					stream.read_partial(16*1024).clear until stream.eof?
+				end.to limit_allocations(size: 100*1024)
+			end
+		end
 	end
 
 	describe '#write' do