Fix psubscribe, unsubscribe, punsubscribe

- Spawn subscription loop with Threads.@Spawn instead of @async
- Add 'key' field to psubscribe struct
- Make callback functions in subscription accept 1 argument: SubscriptionMessage, instead of just SubscriptionMessage.message
- Lock pack_command to prevent race conditions writing to the same socket

Author: captchanjack

Project.toml

@@ -1,6 +1,6 @@
 name = "Redis"
 uuid = "0cf705f9-a9e2-50d1-a699-2b372a39b750"
-version = "1.0.0"
+version = "1.1.0"
 
 [deps]
 DataStructures = "864edb3b-99cc-5e75-8d2d-829cb0a9cfe8"

README.md

@@ -129,7 +129,7 @@ Multiple channels can be subscribed together by providing a `Dict{String, Functi
 
 ```julia
 x = Any[]
-f(y) = push!(x, y)
+f(y::SubscriptionMessage) = push!(x, y)
 sub = open_subscription(conn)
 d = Dict{String, Function}({"baz" => f, "bar" => println})
 subscribe(sub, d)
@@ -140,7 +140,7 @@ publish(conn, "bar", "anything") # "anything" written to stdout
 
 Pattern subscription works in the same way through use of the `psubscribe` function. Channels can be unsubscribed through `unsubscribe` and `punsubscribe`.
 
-Note that the async event loop currently runs until the `SubscriptionConnection` is disconnected, regardless of how many subscriptions the client has active. Event loop error handling should be improved in an update to the API.
+Note that the event loop spawned with Threads.@spawn currently runs until the `SubscriptionConnection` is disconnected, regardless of how many subscriptions the client has active. Event loop error handling should be improved in an update to the API.
 
 ### Subscription error handling
 

src/client.jl

@@ -145,7 +145,7 @@ end
 nullcb(err) = @debug err
 function open_subscription(conn::RedisConnection, err_callback=nullcb)
     s = SubscriptionConnection(conn)
-    @async subscription_loop(s, err_callback)
+    Threads.@spawn subscription_loop(s, err_callback)
     s
 end
 
@@ -157,9 +157,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.message)
+                conn.callbacks[message.channel](message)
             elseif message.message_type == SubscriptionMessageType.Pmessage
-                conn.pcallbacks[message.channel](message.message)
+                conn.pcallbacks[message.channel](message)
             end
         catch err
             err_callback(err)

src/commands.jl

@@ -304,7 +304,7 @@ function unsubscribe(conn::SubscriptionConnection, channels...)
         delete!(conn.callbacks, channel)
     end
     for channel in channels
-        execute_command_without_reply(conn, unshift!([channel], "unsubscribe"))
+        execute_command_without_reply(conn, pushfirst!([channel], "unsubscribe"))
     end
 end
 # function unsubscribe(conn::SubscriptionConnection, channels...)
@@ -319,13 +319,13 @@ function _psubscribe(conn::SubscriptionConnection, patterns::Array)
 end
 
 function psubscribe(conn::SubscriptionConnection, pattern::AbstractString, callback::Function)
-    conn.callbacks[pattern] = callback
+    conn.pcallbacks[pattern] = callback
     _psubscribe(conn, [pattern])
 end
 
 function psubscribe(conn::SubscriptionConnection, subs::Dict{AbstractString, Function})
     for (pattern, callback) in subs
-        conn.callbacks[pattern] = callback
+        conn.pcallbacks[pattern] = callback
     end
     _psubscribe(conn, collect(values(subs)))
 end
@@ -334,7 +334,9 @@ function punsubscribe(conn::SubscriptionConnection, patterns...)
     for pattern in patterns
         delete!(conn.pcallbacks, pattern)
     end
-    execute_command(conn, pushfirst!(patterns, "punsubscribe"))
+    for pattern in patterns
+        execute_command_without_reply(conn, pushfirst!([pattern], "unsubscribe"))
+    end
 end
 
 #Need a specialized version of execute to keep the connection in the transaction state

src/parser.jl

@@ -88,7 +88,9 @@ end
 
 function execute_command_without_reply(conn::RedisConnectionBase, command)
     is_connected(conn) || throw(ConnectionException("Socket is disconnected"))
-    pack_command(conn.socket, command)
+    lock(conn.socket.lock) do 
+        pack_command(conn.socket, command)
+    end
 end
 
 function execute_command(conn::RedisConnectionBase, command::Vector)
@@ -105,15 +107,15 @@ end
 struct SubscriptionMessage
     message_type
     channel::AbstractString
+    key::Union{AbstractString,Nothing}
     message::AbstractString
 
     function SubscriptionMessage(reply::AbstractArray)
-        notification = reply
-        message_type = notification[1]
+        message_type = reply[1]
         if message_type == "message"
-            new(SubscriptionMessageType.Message, notification[2], notification[3])
+            new(SubscriptionMessageType.Message, reply[2], nothing, reply[3])
         elseif message_type == "pmessage"
-            new(SubscriptionMessageType.Pmessage, notification[2], notification[4])
+            new(SubscriptionMessageType.Pmessage, reply[2], reply[3], reply[4])
         else
             new(SubscriptionMessageType.Other, "", "")
         end