pipe: allow half-duplex operation

base/stream.jl

@@ -671,16 +671,16 @@ function uv_readcb(handle::Ptr{Cvoid}, nread::Cssize_t, buf::Ptr{Cvoid})
             if nread == UV_ENOBUFS && nrequested == 0
                 # remind the client that stream.buffer is full
                 notify(stream.cond)
-            elseif nread == UV_EOF # libuv called uv_stop_reading already
+            elseif nread == UV_EOF
+                # n.b. libuv called uv_stop_reading already
                 if stream.status != StatusClosing
                     stream.status = StatusEOF
                     notify(stream.cond)
                     if stream isa TTY
                         # stream can still be used by reseteof (or possibly write)
-                    elseif !(stream isa PipeEndpoint) && ccall(:uv_is_writable, Cint, (Ptr{Cvoid},), stream.handle) != 0
-                        # stream can still be used by write
-                    else
-                        # underlying stream is no longer useful: begin finalization
+                    elseif iszero(ccall(:uv_is_writable, Cint, (Ptr{Cvoid},), stream.handle))
+                        # stream cannot be used by write, so underlying stream
+                        # is no longer useful: do early finalization now
                         ccall(:jl_close_uv, Cvoid, (Ptr{Cvoid},), stream.handle)
                         stream.status = StatusClosing
                     end

stdlib/Sockets/test/runtests.jl

@@ -602,6 +602,43 @@ end
     end
 end
 
+# Test that closewrite (SHUT_WR) works for PipeEndpoint too
+@testset "closewrite half-duplex" begin
+    mktempdir() do dir
+        socketname = Sys.iswindows() ? ("\\\\.\\pipe\\uv-test-" * randstring(6)) : joinpath(dir, "s.sock")
+        server = listen(socketname)
+        t = @async begin
+            c = accept(server)
+            try
+                close(server)
+                @test isreadable(c)
+                @test iswritable(c)
+                @test read(c, String) == "ping"
+                @test !isreadable(c)
+                sleep(0.05)
+                write(c, "pong")
+                @test iswritable(c)
+                :ok
+            finally
+                close(c)
+            end
+        end
+        client = connect(socketname)
+        write(client, "ping")
+        @test isreadable(client)
+        @test iswritable(client)
+        closewrite(client)
+        @test isreadable(client)
+        @test !iswritable(client)
+        out = read(client, String)
+        @test !isreadable(client)
+        close(client)
+        srv = fetch(t)
+        @test out == "pong"
+        @test srv === :ok
+    end
+end
+
 @testset "TCPSocket RawFD constructor" begin
     if Sys.islinux()
         let fd = ccall(:socket, Int32, (Int32, Int32, Int32),