lib/redis/subscribe.rb in redis-2.0.0.rc1 vs lib/redis/subscribe.rb in redis-2.0.0.rc2

@@ -2,53 +2,43 @@
   class SubscribedClient
     def initialize(client)
       @client = client
     end
 
-    def call(command, *args)
-      @client.call_async(command, *args)
+    def call(*args)
+      @client.process(args)
     end
 
     def subscribe(*channels, &block)
-      @client.call_async(:subscribe, *channels)
+      subscription("subscribe", "unsubscribe", channels, block)
+    end
 
-      sub = Subscription.new(&block)
-
-      begin
-        loop do
-          type, channel, message = @client.read
-          sub.callbacks[type].call(channel, message)
-          break if type == "unsubscribe" && message == 0
-        end
-      ensure
-        @client.call_async(:unsubscribe)
-      end
+    def psubscribe(*channels, &block)
+      subscription("psubscribe", "punsubscribe", channels, block)
     end
 
     def unsubscribe(*channels)
-      @client.call_async(:unsubscribe, *channels)
-      @client
+      call(:unsubscribe, *channels)
     end
 
-    def psubscribe(*channels, &block)
-      @client.call_async(:psubscribe, *channels)
+    def punsubscribe(*channels)
+      call(:punsubscribe, *channels)
+    end
 
+  protected
+
+    def subscription(start, stop, channels, block)
       sub = Subscription.new(&block)
 
       begin
-        loop do
-          type, pattern, channel, message = @client.read
-          sub.callbacks[type].call(pattern, channel, message)
-          break if type == "punsubscribe" && channel == 0
+        @client.call_loop(start, *channels) do |line|
+          type, *rest = line
+          sub.callbacks[type].call(*rest)
+          break if type == stop && rest.last == 0
         end
       ensure
-        @client.call_async(:punsubscribe)
+        send(stop)
       end
-    end
-
-    def punsubscribe(*channels)
-      @client.call_async(:punsubscribe, *channels)
-      @client
     end
   end
 
   class Subscription
     attr :callbacks