Merge pull request #102 from JuliaDatabases/tan/psub

fix: psubscribe method issues

Author: jkaye2012

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, subscribe_data, publish, psubscribe, punsubscribe, unsubscribe
+export subscribe, subscribe_data, publish, psubscribe, psubscribe_data, 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

@@ -287,35 +287,46 @@ function _subscribe(conn::SubscriptionConnection, channels::Array)
 end
 
 @enum CallbackType  CallbackMsgStruct = 1 CallbackMsgData = 2
+@enum SubscriptionType SubscriptionPlain = 1 SubscriptionPattern = 2
+
+function _register_callback(
+    conn::SubscriptionConnection,
+    channel::AbstractString,
+    subscription_type::SubscriptionType,
+    callback::Function,
+    callback_type::CallbackType,
+)
+    callbacks = subscription_type == SubscriptionPlain ? conn.callbacks :
+        subscription_type == SubscriptionPattern ? conn.pcallbacks :
+        error("Invalid callback type")
 
-function _register_callback(conn::SubscriptionConnection, channel::AbstractString, callback::Function, callback_type::CallbackType)
     if callback_type == CallbackMsgStruct
-        conn.callbacks[channel] = callback
+        callbacks[channel] = callback
     elseif callback_type == CallbackMsgData
-        conn.callbacks[channel] = (msg_struct)->callback(msg_struct.message)
+        callbacks[channel] = (msg_struct)->callback(msg_struct.message)
     end
 end
 
 function subscribe(conn::SubscriptionConnection, channel::AbstractString, callback::Function)
-    _register_callback(conn, channel, callback, CallbackMsgStruct)
+    _register_callback(conn, channel, SubscriptionPlain, callback, CallbackMsgStruct)
     _subscribe(conn, [channel])
 end
 
-function subscribe(conn::SubscriptionConnection, subs::Dict{AbstractString, Function})
+function subscribe(conn::SubscriptionConnection, subs::Dict)
     for (channel, callback) in subs
-        _register_callback(conn, channel, callback, CallbackMsgStruct)
+        _register_callback(conn, channel, SubscriptionPlain, callback, CallbackMsgStruct)
     end
     _subscribe(conn, collect(keys(subs)))
 end
 
 function subscribe_data(conn::SubscriptionConnection, channel::AbstractString, callback::Function)
-    _register_callback(conn, channel, callback, CallbackMsgData)
+    _register_callback(conn, channel, SubscriptionPlain, callback, CallbackMsgData)
     _subscribe(conn, [channel])
 end
 
-function subscribe_data(conn::SubscriptionConnection, subs::Dict{AbstractString, Function})
+function subscribe_data(conn::SubscriptionConnection, subs::Dict)
     for (channel, callback) in subs
-        _register_callback(conn, channel, callback, CallbackMsgData)
+        _register_callback(conn, channel, SubscriptionPlain, callback, CallbackMsgData)
     end
     _subscribe(conn, collect(keys(subs)))
 end
@@ -341,15 +352,27 @@ function _psubscribe(conn::SubscriptionConnection, patterns::Array)
 end
 
 function psubscribe(conn::SubscriptionConnection, pattern::AbstractString, callback::Function)
-    conn.pcallbacks[pattern] = callback
+    _register_callback(conn, pattern, SubscriptionPattern, callback, CallbackMsgStruct)
     _psubscribe(conn, [pattern])
 end
 
-function psubscribe(conn::SubscriptionConnection, subs::Dict{AbstractString, Function})
+function psubscribe(conn::SubscriptionConnection, subs::Dict)
     for (pattern, callback) in subs
-        conn.pcallbacks[pattern] = callback
+        _register_callback(conn, pattern, SubscriptionPattern, callback, CallbackMsgStruct)
     end
-    _psubscribe(conn, collect(values(subs)))
+    _psubscribe(conn, collect(keys(subs)))
+end
+
+function psubscribe_data(conn::SubscriptionConnection, pattern::AbstractString, callback::Function)
+    _register_callback(conn, pattern, SubscriptionPattern, callback, CallbackMsgData)
+    _psubscribe(conn, [pattern])
+end
+
+function psubscribe_data(conn::SubscriptionConnection, subs::Dict)
+    for (pattern, callback) in subs
+        _register_callback(conn, pattern, SubscriptionPattern, callback, CallbackMsgData)
+    end
+    _psubscribe(conn, collect(keys(subs)))
 end
 
 function punsubscribe(conn::SubscriptionConnection, patterns...)

test/redis_tests.jl

@@ -395,18 +395,44 @@ end
         @warn "Error while processing subscription: $err"
         return nothing
     end
-    x = Any[]
-    function f(y::AbstractString)
-        push!(x, y)
+
+    string_matched_results = Any[]
+    pattern_matched_results = Any[]
+
+    function string_matched(y::AbstractString)
+        push!(string_matched_results, y)
+    end
+
+    function pattern_matched(y::AbstractString)
+        push!(pattern_matched_results, y)
     end
+
     subs = open_subscription(conn, handleException) #handleException is called when an exception occurs
-    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
-    @test publish(conn, "duplicate", "hello world 2") > 0 #Number of connected clients returned
-    sleep(2)
-    @test x == ["hello, world!", "Okay, bye!", "hello world 2"]
+
+    subscribe_data(subs, "1channel", string_matched)
+    psubscribe_data(subs, "1cha??el", pattern_matched)
+
+    subscribe(subs, "2channel", y->string_matched(y.message))
+    psubscribe(subs, "2chan*", y->pattern_matched(y.message))
+
+    subscribe_data(subs, Dict("3channel" => string_matched))
+    psubscribe_data(subs, Dict("3cha??el" => pattern_matched))
+
+    subscribe(subs, Dict("4channel" => y->string_matched(y.message)))
+    psubscribe(subs, Dict("4chan*" => y->pattern_matched(y.message)))
+
+    published_messages = ["hello, world1!", "hello, world 2!", "hello, world 3!", "hello, world 4!"]
+
+    for idx in 1:4
+        @test publish(conn, string(idx)*"channel", published_messages[idx]) > 0 #Number of connected clients returned
+    end
+
+    timedwait(5.0; pollint=1.0) do # wait for the messages to be received
+        length(string_matched_results) == 4 && length(pattern_matched_results) == 4        
+    end
+
+    @test string_matched_results == published_messages
+    @test pattern_matched_results == published_messages
 
     # following command prints ("Invalid response received: ")
     disconnect(subs)