Merge pull request #100 from JuliaDatabases/tan/subs

fix: avoid worldage issue in subscription callback

Author: jkaye2012

README.md

@@ -133,7 +133,7 @@ Multiple channels can be subscribed together by providing a `Dict{String, Functi
 x = Any[]
 f(y::SubscriptionMessage) = push!(x, y.message)
 sub = open_subscription(conn)
-d = Dict{String, Function}({"baz" => f, "bar" => y->println(y.message)})
+d = Dict{String, Function}("baz" => f, "bar" => y->println(y.message))
 subscribe(sub, d)
 publish(conn, "baz", "foobar")
 x # Returns ["foobar"]

src/client.jl

@@ -165,9 +165,9 @@ function subscription_loop(conn::SubscriptionConnection, err_callback::Function)
             reply = convert_reply(reply)
             message = SubscriptionMessage(reply)
             if message.message_type == SubscriptionMessageType.Message
-                conn.callbacks[message.channel](message)
+                Base.invokelatest(conn.callbacks[message.channel], message)
             elseif message.message_type == SubscriptionMessageType.Pmessage
-                conn.pcallbacks[message.channel](message)
+                Base.invokelatest(conn.pcallbacks[message.channel], message)
             end
         catch err
             err_callback(err)

test/redis_tests.jl

@@ -404,8 +404,9 @@ end
     subscribe(subs, "duplicate", y->f(y.message))
     @test publish(conn, "channel", "hello, world!") > 0 #Number of connected clients returned
     @test publish(conn, "channel", "Okay, bye!") > 0 #Number of connected clients returned
+    @test publish(conn, "duplicate", "hello world 2") > 0 #Number of connected clients returned
     sleep(2)
-    @test x == ["hello, world!", "Okay, bye!"]
+    @test x == ["hello, world!", "Okay, bye!", "hello world 2"]
 
     # following command prints ("Invalid response received: ")
     disconnect(subs)