Fix compatibility with redis-rb 4.6.0

Fix: resque#191
Fix: redis/redis-rb#1088

In redis-rb 4.6.0, the object yielded by `multi` and `pipelined`
is an unsynchronized `PipelinedConnection` object.

This changed opened a thread safety issue in Redis::Namespace:

  - T1: `redis.multi do`, sets `Namespace#redis = PipelinedConnection`
  - T2: any command called before T1 exists the `multi` block is called on the pipelined connection

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## master
 
+- Fix compatibility with redis-rb 4.6.0. `Redis::Namespace#multi` and `Redis::Namespace#pipelined` were no longer
+  thread-safe. Calling these methods concurrently onthe same instance could cause pipelines or transaction to be
+  intertwined. See https://github.com/resque/redis-namespace/issues/191 and https://github.com/redis/redis-rb/issues/1088
+
 ## 1.8.1
 
  - Allow Ruby 3.0 version in gemspec

lib/redis/namespace.rb

@@ -492,6 +492,12 @@ def call_with_namespace(command, *args, &block)
     end
     ruby2_keywords(:call_with_namespace) if respond_to?(:ruby2_keywords, true)
 
+  protected
+
+    def redis=(redis)
+      @redis = redis
+    end
+
   private
 
     if Hash.respond_to?(:ruby2_keywords_hash)
@@ -521,12 +527,9 @@ def call_site
 
     def namespaced_block(command, &block)
       redis.send(command) do |r|
-        begin
-          original, @redis = @redis, r
-          yield self
-        ensure
-          @redis = original
-        end
+        copy = dup
+        copy.redis = r
+        yield copy
       end
     end
 

spec/redis_spec.rb

@@ -6,12 +6,9 @@
   @redis_version = Gem::Version.new(Redis.current.info["redis_version"])
   let(:redis_client) { @redis.respond_to?(:_client) ? @redis._client : @redis.client}
 
-  before(:all) do
+  before(:each) do
     # use database 15 for testing so we dont accidentally step on your real data
     @redis = Redis.new :db => 15
-  end
-
-  before(:each) do
     @namespaced = Redis::Namespace.new(:ns, :redis => @redis)
     @redis.flushdb
     @redis.set('foo', 'bar')
@@ -398,6 +395,27 @@
     expect(result).to eq(["bar", "value"])
   end
 
+  it "is thread safe for multi blocks" do
+    mon = Monitor.new
+    entered = false
+    entered_cond = mon.new_cond
+
+    thread = Thread.new do
+      mon.synchronize do
+        entered_cond.wait_until { entered }
+        @namespaced.multi
+      end
+    end
+
+    @namespaced.multi do |transaction|
+      entered = true
+      mon.synchronize { entered_cond.signal }
+      thread.join(0.1)
+      transaction.get("foo")
+    end
+    thread.join
+  end
+
   it "should add namespace to strlen" do
     @namespaced.set("mykey", "123456")
     expect(@namespaced.strlen("mykey")).to eq(6)