Merge pull request #92 from BambOoxX/generalize-serialization

Implement JSON serialization interface

Author: davidanthoff

src/core.jl

@@ -112,7 +112,7 @@ struct Request
     token::Union{CancellationTokens.CancellationToken,Nothing}
 end
 
-mutable struct JSONRPCEndpoint{IOIn <: IO,IOOut <: IO}
+mutable struct JSONRPCEndpoint{IOIn<:IO,IOOut<:IO,S<:JSON.Serialization}
     pipe_in::IOIn
     pipe_out::IOOut
 
@@ -129,9 +129,11 @@ mutable struct JSONRPCEndpoint{IOIn <: IO,IOOut <: IO}
 
     read_task::Union{Nothing,Task}
     write_task::Union{Nothing,Task}
+
+    serialization::S
 end
 
-JSONRPCEndpoint(pipe_in, pipe_out, err_handler = nothing) =
+JSONRPCEndpoint(pipe_in, pipe_out, err_handler=nothing, serialization::JSON.Serialization=JSON.StandardSerialization()) =
     JSONRPCEndpoint(
         pipe_in,
         pipe_out,
@@ -143,7 +145,8 @@ JSONRPCEndpoint(pipe_in, pipe_out, err_handler = nothing) =
         err_handler,
         :idle,
         nothing,
-        nothing)
+        nothing,
+        serialization)
 
 function write_transport_layer(stream, response)
     response_utf8 = transcode(UInt8, response)
@@ -290,7 +293,7 @@ function send_notification(x::JSONRPCEndpoint, method::AbstractString, params)
 
     message = Dict("jsonrpc" => "2.0", "method" => method, "params" => params)
 
-    message_json = JSON.json(message)
+    message_json = sprint(JSON.show_json, x.serialization, message)
 
     put!(x.out_msg_queue, message_json)
 
@@ -306,7 +309,7 @@ function send_request(x::JSONRPCEndpoint, method::AbstractString, params)
     response_channel = Channel{Any}(1)
     x.outstanding_requests[id] = response_channel
 
-    message_json = JSON.json(message)
+    message_json = sprint(JSON.show_json, x.serialization, message)
 
     put!(x.out_msg_queue, message_json)
 
@@ -355,7 +358,7 @@ function send_success_response(endpoint, original_request::Request, result)
 
     response = Dict("jsonrpc" => "2.0", "id" => original_request.id, "result" => result)
 
-    response_json = JSON.json(response)
+    response_json = sprint(JSON.show_json, endpoint.serialization, response)
 
     put!(endpoint.out_msg_queue, response_json)
 end
@@ -369,7 +372,7 @@ function send_error_response(endpoint, original_request::Request, code, message,
 
     response = Dict("jsonrpc" => "2.0", "id" => original_request.id, "error" => Dict("code" => code, "message" => message, "data" => data))
 
-    response_json = JSON.json(response)
+    response_json = sprint(JSON.show_json, endpoint.serialization, response)
 
     put!(endpoint.out_msg_queue, response_json)
 end

test/runtests.jl

@@ -1,3 +1,4 @@
 using TestItemRunner
 
 @run_package_tests
+    

test/shared_test_code.jl

@@ -18,3 +18,35 @@
     Base.:(==)(a::Foo2,b::Foo2) = a.fieldA == b.fieldA && a.fieldB == b.fieldB
 
 end
+
+@testmodule NamedPipes begin
+    using Sockets, JSONRPC
+
+    export get_named_pipe
+
+    function get_named_pipe()
+        socket_name = JSONRPC.generate_pipe_name()
+
+        server_is_up = Channel(1)
+
+        socket1_channel = Channel(1)
+
+        @async try
+            server = listen(socket_name)
+            put!(server_is_up, true)
+            socket1 = accept(server)
+
+            put!(socket1_channel, socket1)
+        catch err
+            Base.display_error(err, catch_backtrace())
+        end
+
+        wait(server_is_up)
+
+        socket2 = connect(socket_name)
+
+        socket1 = take!(socket1_channel)
+        
+        return socket1, socket2
+    end
+end

test/test_json_serialization.jl

@@ -0,0 +1,75 @@
+@testitem "Custom JSON serialization" setup=[NamedPipes] begin
+    using JSON
+    using JSON: StructuralContext, begin_object, show_pair, end_object, show_json, Serializations.StandardSerialization
+
+    struct OurSerialization <: JSON.Serializations.CommonSerialization end
+
+    struct OurStruct
+        a::String
+        b::String
+    end
+
+    function JSON.show_json(io::StructuralContext, s::OurSerialization, f::OurStruct)
+        show_json(io, StandardSerialization(), "$(f.a):$(f.b)")
+    end
+
+    x = OurStruct("Hello", "World")
+
+    socket1, socket2 = NamedPipes.get_named_pipe()
+
+    task_done = Channel(1)
+
+    messages_back = Channel(Inf)
+
+    @async try
+        ep2 = JSONRPCEndpoint(socket1, socket1, nothing, OurSerialization())
+        
+        run(ep2)
+
+        msg = JSONRPC.get_next_message(ep2)
+        put!(messages_back, msg)
+
+        msg2 = JSONRPC.get_next_message(ep2)
+        put!(messages_back, msg2)
+        send_success_response(ep2, msg2, [x])
+
+        msg3 = JSONRPC.get_next_message(ep2)
+        put!(messages_back, msg3)
+        send_error_response(ep2, msg3, 5, "Error", [x])
+
+        close(ep2)
+    finally
+        put!(task_done, true)
+    catch err
+        Base.display_error(err, catch_backtrace())
+    end   
+
+    ep1 = JSONRPCEndpoint(socket2, socket2, nothing, OurSerialization())
+
+    run(ep1)
+
+    send_notification(ep1, "foo", [x])
+
+    response1 = send_request(ep1, "bar", [x])
+    try
+        send_request(ep1, "bar", [x])
+    catch err_msg
+        if err_msg isa JSONRPC.JSONRPCError
+            @test err_msg.data == Any["Hello:World"]
+        else
+            rethrow(err_msg)
+        end
+    end
+
+    close(ep1)
+    
+    wait(task_done)
+
+    msg1 = take!(messages_back)
+    msg2 = take!(messages_back)
+    msg3 = take!(messages_back)
+
+    @test msg1.params == ["Hello:World"]
+    @test msg2.params == ["Hello:World"]
+    @test msg3.params == ["Hello:World"]
+end