Handle closed pipes better

Author: davidanthoff

src/core.jl

@@ -70,20 +70,28 @@ function write_transport_layer(stream, response)
 end
 
 function read_transport_layer(stream)
-    header_dict = Dict{String,String}()
-    line = chomp(readline(stream))
-    # Check whether the socket was closed
-    if line == ""
-        return nothing
-    end
-    while length(line) > 0
-        h_parts = split(line, ":")
-        header_dict[chomp(h_parts[1])] = chomp(h_parts[2])
+    try
+        header_dict = Dict{String,String}()
         line = chomp(readline(stream))
+        # Check whether the socket was closed
+        if line == ""
+            return nothing
+        end
+        while length(line) > 0
+            h_parts = split(line, ":")
+            header_dict[chomp(h_parts[1])] = chomp(h_parts[2])
+            line = chomp(readline(stream))
+        end
+        message_length = parse(Int, header_dict["Content-Length"])
+        message_str = String(read(stream, message_length))
+        return message_str
+    catch err
+        if err isa Base.IOError
+            return nothing
+        end
+
+        rethrow(err)
     end
-    message_length = parse(Int, header_dict["Content-Length"])
-    message_str = String(read(stream, message_length))
-    return message_str
 end
 
 Base.isopen(x::JSONRPCEndpoint) = x.status != :closed && isopen(x.pipe_in) && isopen(x.pipe_out)
@@ -114,41 +122,41 @@ function Base.run(x::JSONRPCEndpoint)
     end
 
     x.read_task = @async try
-        try
-            while true
-                message = read_transport_layer(x.pipe_in)
+        while true
+            message = read_transport_layer(x.pipe_in)
 
-                if message === nothing || x.status == :closed
-                    break
-                end
+            if message === nothing || x.status == :closed
+                break
+            end
 
-                message_dict = JSON.parse(message)
-
-                if haskey(message_dict, "method")
-                    try
-                        put!(x.in_msg_queue, message_dict)
-                    catch err
-                        if err isa InvalidStateException
-                            break
-                        else
-                            rethrow(err)
-                        end
-                    end
-                else
-                    # This must be a response
-                    id_of_request = message_dict["id"]
+            message_dict = JSON.parse(message)
 
-                    channel_for_response = x.outstanding_requests[id_of_request]
-                    put!(channel_for_response, message_dict)
+            if haskey(message_dict, "method")
+                try
+                    put!(x.in_msg_queue, message_dict)
+                catch err
+                    if err isa InvalidStateException
+                        break
+                    else
+                        rethrow(err)
+                    end
                 end
+            else
+                # This must be a response
+                id_of_request = message_dict["id"]
+
+                channel_for_response = x.outstanding_requests[id_of_request]
+                put!(channel_for_response, message_dict)
             end
-        finally
-            close(x.in_msg_queue)
         end
+        
+        close(x.in_msg_queue)
 
         for i in values(x.outstanding_requests)
             close(i)
         end
+
+        c.status = :closed
     catch err
         bt = catch_backtrace()
         if x.err_handler !== nothing