Merge pull request #95 from JuliaDatabases/tan/pubsub

fix subscription callback invoke and pubsub tests

Author: jkaye2012

README.md

@@ -116,11 +116,13 @@ publish(conn, "channel", "hello, world!")
 
 Subscriptions are handled using the `SubscriptionConnection`. Similar to the `TransactionConnection`, the `SubscriptionConnection` is constructed from an existing `RedisConnection`. Once created, the `SubscriptionConnection` maintains a simple event loop that will call the user's defined function whenever a message is received on the specified channel.
 
+If the `subscribe_data` method is used for subscription then the callback function will be passed the `message` field of `SubscriptionMessage` instance. If the `subscribe` method is used for subscription, the callback will be passed a `SubscriptionMessage` directly, which contains the channel, message type and key as well. 
+
 ```julia
 x = Any[]
 f(y) = push!(x, y)
 sub = open_subscription(conn)
-subscribe(sub, "baz", f)
+subscribe_data(sub, "baz", f)
 publish(conn, "baz", "foobar")
 x # Returns ["foobar"]
 ```
@@ -129,9 +131,9 @@ Multiple channels can be subscribed together by providing a `Dict{String, Functi
 
 ```julia
 x = Any[]
-f(y::SubscriptionMessage) = push!(x, y)
+f(y::SubscriptionMessage) = push!(x, y.message)
 sub = open_subscription(conn)
-d = Dict{String, Function}({"baz" => f, "bar" => println})
+d = Dict{String, Function}({"baz" => f, "bar" => y->println(y.message)})
 subscribe(sub, d)
 publish(conn, "baz", "foobar")
 x # Returns ["foobar"]

src/Redis.jl

@@ -45,7 +45,7 @@ export discard, exec, multi, unwatch, watch
 # Scripting commands
 export evalscript, evalsha, script_exists, script_flush, script_kill, script_load
 # PubSub commands
-export subscribe, publish, psubscribe, punsubscribe, unsubscribe
+export subscribe, subscribe_data, publish, psubscribe, punsubscribe, unsubscribe
 # Server commands
 export bgrewriteaof, bgsave, client_list, client_id, client_pause, client_setname, cluster_slots,
        command, command_count, command_info, config_get, config_resetstat, config_rewrite,

src/commands.jl

@@ -286,14 +286,36 @@ function _subscribe(conn::SubscriptionConnection, channels::Array)
     execute_command_without_reply(conn, pushfirst!(channels, "subscribe"))
 end
 
+@enum CallbackType  CallbackMsgStruct = 1 CallbackMsgData = 2
+
+function _register_callback(conn::SubscriptionConnection, channel::AbstractString, callback::Function, callback_type::CallbackType)
+    if callback_type == CallbackMsgStruct
+        conn.callbacks[channel] = callback
+    elseif callback_type == CallbackMsgData
+        conn.callbacks[channel] = (msg_struct)->callback(msg_struct.message)
+    end
+end
+
 function subscribe(conn::SubscriptionConnection, channel::AbstractString, callback::Function)
-    conn.callbacks[channel] = callback
+    _register_callback(conn, channel, callback, CallbackMsgStruct)
     _subscribe(conn, [channel])
 end
 
 function subscribe(conn::SubscriptionConnection, subs::Dict{AbstractString, Function})
     for (channel, callback) in subs
-        conn.callbacks[channel] = callback
+        _register_callback(conn, channel, callback, CallbackMsgStruct)
+    end
+    _subscribe(conn, collect(keys(subs)))
+end
+
+function subscribe_data(conn::SubscriptionConnection, channel::AbstractString, callback::Function)
+    _register_callback(conn, channel, callback, CallbackMsgData)
+    _subscribe(conn, [channel])
+end
+
+function subscribe_data(conn::SubscriptionConnection, subs::Dict{AbstractString, Function})
+    for (channel, callback) in subs
+        _register_callback(conn, channel, callback, CallbackMsgData)
     end
     _subscribe(conn, collect(keys(subs)))
 end

test/redis_tests.jl

@@ -396,12 +396,12 @@ end
         return nothing
     end
     x = Any[]
-    function f(y::String)
+    function f(y::AbstractString)
         push!(x, y)
     end
     subs = open_subscription(conn, handleException) #handleException is called when an exception occurs
-    subscribe(subs, "channel", f)
-    subscribe(subs, "duplicate", f)
+    subscribe_data(subs, "channel", f)
+    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
     sleep(2)