implement LibuvStream interface for BufferStream

base/stream.jl

@@ -23,14 +23,14 @@ abstract type LibuvStream <: IO end
 # .  +- Pipe
 # .  +- Process (not exported)
 # .  +- ProcessChain (not exported)
-# +- BufferStream
 # +- DevNull (not exported)
 # +- Filesystem.File
 # +- LibuvStream (not exported)
 # .  +- PipeEndpoint (not exported)
 # .  +- TCPSocket
 # .  +- TTY (not exported)
 # .  +- UDPSocket
+# .  +- BufferStream (FIXME: 2.0)
 # +- IOBuffer = Base.GenericIOBuffer{Array{UInt8,1}}
 # +- IOStream
 
@@ -1202,11 +1202,12 @@ end
 mutable struct BufferStream <: LibuvStream
     buffer::IOBuffer
     cond::Threads.Condition
+    readerror::Any
     is_open::Bool
     buffer_writes::Bool
     lock::ReentrantLock # advisory lock
 
-    BufferStream() = new(PipeBuffer(), Threads.Condition(), true, false, ReentrantLock())
+    BufferStream() = new(PipeBuffer(), Threads.Condition(), nothing, true, false, ReentrantLock())
 end
 
 isopen(s::BufferStream) = s.is_open

test/iobuffer.jl

@@ -254,6 +254,18 @@ end
     @test close(bstream) === nothing
     @test eof(bstream)
     @test bytesavailable(bstream) == 0
+    flag = Ref{Bool}(false)
+    event = Base.Event()
+    bstream = Base.BufferStream()
+    task = @async begin
+        notify(event)
+        read(bstream, 16)
+        flag[] = true
+    end
+    wait(event)
+    write(bstream, rand(UInt8, 16))
+    wait(task)
+    @test flag[] == true
 end
 
 @test flush(IOBuffer()) === nothing # should be a no-op