Merge pull request #54 from julia-vscode/better-error-handling

Improve error handling in @async

Author: pfitzseb

src/core.jl

@@ -91,13 +91,17 @@ function Base.run(x::JSONRPCEndpoint)
     x.status == :idle || error("Endpoint is not idle.")
 
     x.write_task = @async try
-        for msg in x.out_msg_queue
-            if isopen(x.pipe_out)
-                write_transport_layer(x.pipe_out, msg)
-            else
-                # TODO Reconsider at some point whether this should be treated as an error.
-                break
+        try
+            for msg in x.out_msg_queue
+                if isopen(x.pipe_out)
+                    write_transport_layer(x.pipe_out, msg)
+                else
+                    # TODO Reconsider at some point whether this should be treated as an error.
+                    break
+                end
             end
+        finally
+            close(x.out_msg_queue)
         end
     catch err
         bt = catch_backtrace()
@@ -106,37 +110,39 @@ function Base.run(x::JSONRPCEndpoint)
         else
             Base.display_error(stderr, err, bt)
         end
-    finally
-        close(x.out_msg_queue)
     end
 
     x.read_task = @async try
-        while true
-            message = read_transport_layer(x.pipe_in)
-
-            if message === nothing || x.status == :closed
-                break
-            end
+        try
+            while true
+                message = read_transport_layer(x.pipe_in)
 
-            message_dict = JSON.parse(message)
+                if message === nothing || x.status == :closed
+                    break
+                end
 
-            if haskey(message_dict, "method")
-                try
-                    put!(x.in_msg_queue, message_dict)
-                catch err
-                    if err isa InvalidStateException
-                        break
-                    else
-                        rethrow(err)
+                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
-                end
-            else
-                # This must be a response
-                id_of_request = message_dict["id"]
+                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)
+                    channel_for_response = x.outstanding_requests[id_of_request]
+                    put!(channel_for_response, message_dict)
+                end
             end
+        finally
+            close(x.in_msg_queue)
         end
     catch err
         bt = catch_backtrace()
@@ -145,8 +151,6 @@ function Base.run(x::JSONRPCEndpoint)
         else
             Base.display_error(stderr, err, bt)
         end
-    finally
-        close(x.in_msg_queue)
     end
 
     x.status = :running

test/test_typed.jl

@@ -1,9 +1,9 @@
 @testset "Message dispatcher" begin
 
     if Sys.iswindows()
-        global_socket_name = "\\\\.\\pipe\\jsonrpc-testrun"
+        global_socket_name1 = "\\\\.\\pipe\\jsonrpc-testrun1"
     elseif Sys.isunix()
-        global_socket_name = joinpath(tempdir(), "jsonrpc-testrun")
+        global_socket_name1 = joinpath(tempdir(), "jsonrpc-testrun1")
     else
         error("Unknown operating system.")
     end
@@ -16,8 +16,8 @@
 
     server_is_up = Base.Condition()
 
-    server_task = @async begin
-        server = listen(global_socket_name)
+    server_task = @async try
+        server = listen(global_socket_name1)
         notify(server_is_up)
         sock = accept(server)
         global conn = JSONRPC.JSONRPCEndpoint(sock, sock)
@@ -35,11 +35,13 @@
         for msg in conn
             JSONRPC.dispatch_msg(conn, msg_dispatcher, msg)
         end
+    catch err
+        Base.display_error(stderr, err, catch_backtrace())
     end
 
     wait(server_is_up)
 
-    sock2 = connect(global_socket_name)
+    sock2 = connect(global_socket_name1)
     conn2 = JSONRPCEndpoint(sock2, sock2)
 
     run(conn2)
@@ -61,10 +63,18 @@
 
     # Now we test a faulty server
 
+    if Sys.iswindows()
+        global_socket_name2 = "\\\\.\\pipe\\jsonrpc-testrun2"
+    elseif Sys.isunix()
+        global_socket_name2 = joinpath(tempdir(), "jsonrpc-testrun2")
+    else
+        error("Unknown operating system.")
+    end
+
     server_is_up = Base.Condition()
 
-    server_task2 = @async begin
-        server = listen(global_socket_name)
+    server_task2 = @async try
+        server = listen(global_socket_name2)
         notify(server_is_up)
         sock = accept(server)
         global conn = JSONRPC.JSONRPCEndpoint(sock, sock)
@@ -77,11 +87,13 @@
         for msg in conn
             @test_throws ErrorException("The handler for the 'request2' request returned a value of type $Int, which is not a valid return type according to the request definition.") JSONRPC.dispatch_msg(conn, msg_dispatcher, msg)
         end
+    catch err
+        Base.display_error(stderr, err, catch_backtrace())
     end
 
     wait(server_is_up)
 
-    sock2 = connect(global_socket_name)
+    sock2 = connect(global_socket_name2)
     conn2 = JSONRPCEndpoint(sock2, sock2)
 
     run(conn2)