Merge pull request #4 from staticfloat/sf/block_dont_error

Don't throw errors if `chunk` is larger than `bs.max_len`

Author: staticfloat

src/BufferStream.jl

@@ -53,23 +53,29 @@ function append_chunk(bs::BufferStream, data::AbstractVector{UInt8})
         throw(ArgumentError("Stream is closed"))
     end
 
-    if bs.max_len != 0 && length(data) > bs.max_len
-        throw(ArgumentError("Chunk is too large to fit into this BufferStream!"))
-    end
-
     # Copy the data so that users can't clobber our internal list
     lock(bs.write_cond) do
-        # If we would exceed our maximum length, then we must wait until someone reads from us.
-        while bs.max_len != 0 && length(bs) + length(data) > bs.max_len
-            wait(bs.write_cond)
+        data_written = 0
+        while data_written < length(data)
+            if bs.max_len == 0
+                space_available = length(data)
+            else
+                space_available = bs.max_len - length(bs)
+            end
+            if space_available == 0
+                wait(bs.write_cond)
+                continue
+            end
+            bytes_to_write = min(space_available, length(data) - data_written)
+            push!(bs.chunks, data[data_written+1:data_written+bytes_to_write])
+            # Notify someone who was waiting for some data
+            lock(bs.read_cond) do
+                notify(bs.read_cond; all=false)
+            end
+            data_written += bytes_to_write
         end
-        push!(bs.chunks, data)
     end
 
-    # Notify someone who was waiting for some data
-    lock(bs.read_cond) do
-        notify(bs.read_cond; all=false)
-    end
     return length(data)
 end
 

test/runtests.jl

@@ -81,7 +81,6 @@ end
 
 @testset "max_len" begin
     bs = BufferStream(1000)
-    @test_throws ArgumentError write(bs, zeros(UInt8, 1001))
     write(bs, zeros(UInt8, 1000))
     t_write = @async begin
         t_elapsed = @elapsed write(bs, zeros(UInt8, 1))
@@ -97,6 +96,20 @@ end
     # Consume the rest of that chunk so it can be dropped, allowing the pending write to go through
     readavailable(bs)
     wait(t_write)
+    # read the last byte that it outputs
+    @test length(readavailable(bs)) == 1
+
+    # Also test writing a chunk larger than the entire buffer
+    t_write = @async begin
+        t_elapsed = @elapsed write(bs, zeros(UInt8, 3000))
+        @test t_elapsed > 0.01
+    end
+    sleep(0.05)
+    # We need to read three times, as each time we can only read 1000 elements
+    @test length(readavailable(bs)) == 1000
+    @test length(readavailable(bs)) == 1000
+    @test length(readavailable(bs)) == 1000
+    wait(t_write)
 end
 
 function tee_task(io_in, io_outs...)