Redis::Client still private. Redis exposes call, queue, commit.

Author: djanowski

lib/redis.rb

@@ -59,6 +59,44 @@ def disconnect!
     @original_client.disconnect
   end
 
+  # Sends a command to Redis and returns its reply.
+  #
+  # Replies are converted to Ruby objects according to the RESP protocol, so
+  # you can expect a Ruby array, integer or nil when Redis sends one. Higher
+  # level transformations, such as converting an array of pairs into a Ruby
+  # hash, are up to consumers.
+  #
+  # Redis error replies are raised as Ruby exceptions.
+  def call(*command)
+    synchronize do |client|
+      client.call(command)
+    end
+  end
+
+  # Queues a command for pipelining.
+  #
+  # Commands in the queue are executed with the Redis::Client#commit method.
+  #
+  # See http://redis.io/topics/pipelining for more details.
+  #
+  def queue(*command)
+    _queue << command
+  end
+
+  # Sends all commands in the queue.
+  #
+  # See http://redis.io/topics/pipelining for more details.
+  #
+  def commit
+    synchronize do |client|
+      begin
+        client.call_pipelined(_queue)
+      ensure
+        _queue.clear
+      end
+    end
+  end
+
   # Authenticate to the server.
   #
   # @param [String] password must match the password specified in the
@@ -2625,6 +2663,10 @@ def _subscription(method, channels, block)
     end
   end
 
+  def _queue
+    Thread.current[:redis_queue] ||= []
+  end
+
 end
 
 require "redis/version"

lib/redis/client.rb

@@ -106,18 +106,8 @@ def initialize(options = {})
       else
         @connector = Connector.new(@options)
       end
-
-      @queue = []
     end
 
-    # Sends a command to Redis and returns its reply.
-    #
-    # Replies are converted to Ruby objects according to the RESP protocol, so
-    # you can expect a Ruby array, integer or nil when Redis sends one. Higher
-    # level transformations, such as converting an array of pairs into a Ruby
-    # hash, are up to consumers.
-    #
-    # Redis error replies are raised as Ruby exceptions.
     def call(command, &block)
       reply = process([command]) { read }
       raise reply if reply.is_a?(CommandError)
@@ -129,26 +119,6 @@ def call(command, &block)
       end
     end
 
-    # Queues a command for pipelining.
-    #
-    # Commands in the queue are executed with the Redis::Client#commit method.
-    #
-    # See http://redis.io/topics/pipelining for more details.
-    #
-    def queue(command)
-      @queue << command
-    end
-
-    # Sends all commands in the queue.
-    #
-    # See http://redis.io/topics/pipelining for more details.
-    #
-    def commit
-      call_pipelined(@queue)
-    ensure
-      @queue.clear
-    end
-
     def connect
       @pid = Process.pid
 

test/client_test.rb

@@ -5,54 +5,54 @@ class TestClient < Test::Unit::TestCase
   include Helper::Client
 
   def test_call
-    result = r.client.call(["PING"])
+    result = r.call("PING")
     assert_equal result, "PONG"
   end
 
   def test_call_with_arguments
-    result = r.client.call(["SET", "foo", "bar"])
+    result = r.call("SET", "foo", "bar")
     assert_equal result, "OK"
   end
 
   def test_call_integers
-    result = r.client.call(["INCR", "foo"])
+    result = r.call("INCR", "foo")
     assert_equal result, 1
   end
 
   def test_call_raise
     assert_raises(Redis::CommandError) do
-      r.client.call(["INCR"])
+      r.call("INCR")
     end
   end
 
   def test_queue_commit
-    r.client.queue(["SET", "foo", "bar"])
-    r.client.queue(["GET", "foo"])
-    result = r.client.commit
+    r.queue("SET", "foo", "bar")
+    r.queue("GET", "foo")
+    result = r.commit
 
     assert_equal result, ["OK", "bar"]
   end
 
   def test_commit_raise
-    r.client.queue(["SET", "foo", "bar"])
-    r.client.queue(["INCR"])
+    r.queue("SET", "foo", "bar")
+    r.queue("INCR")
 
     assert_raise(Redis::CommandError) do
-      r.client.commit
+      r.commit
     end
   end
 
   def test_queue_after_error
-    r.client.queue(["SET", "foo", "bar"])
-    r.client.queue(["INCR"])
+    r.queue("SET", "foo", "bar")
+    r.queue("INCR")
 
     assert_raise(Redis::CommandError) do
-      r.client.commit
+      r.commit
     end
 
-    r.client.queue(["SET",  "foo", "bar"])
-    r.client.queue(["INCR", "baz"])
-    result = r.client.commit
+    r.queue("SET",  "foo", "bar")
+    r.queue("INCR", "baz")
+    result = r.commit
 
     assert_equal result, ["OK", 1]
   end

test/thread_safety_test.rb

@@ -28,5 +28,35 @@ def test_thread_safety
       assert_equal ["1"], $foos.uniq
       assert_equal ["2"], $bars.uniq
     end
+
+    def test_thread_safety_queue_commit
+      redis = Redis.new(OPTIONS)
+      redis.set "foo", 1
+      redis.set "bar", 2
+
+      sample = 100
+
+      t1 = Thread.new do
+        sample.times do
+          r.queue("get", "foo")
+        end
+
+        $foos = r.commit
+      end
+
+      t2 = Thread.new do
+        sample.times do
+          r.queue("get", "bar")
+        end
+
+        $bars = r.commit
+      end
+
+      t1.join
+      t2.join
+
+      assert_equal ["1"], $foos.uniq
+      assert_equal ["2"], $bars.uniq
+    end
   end
 end