Use a thread-local variable for pipeline

Author: lewispb

lib/kredis/migration.rb

@@ -5,7 +5,6 @@ class Kredis::Migration
 
   def initialize(config = :shared)
     @redis = Kredis.configured_for config
-    @pipeline = nil
     # TODO: Replace script loading with `copy` command once Redis 6.2+ is the minimum supported version.
     @copy_sha = @redis.script "load", "redis.call('SETNX', KEYS[2], redis.call('GET', KEYS[1])); return 1;"
   end
@@ -40,10 +39,6 @@ def delete_all(key_pattern)
   private
     SCAN_BATCH_SIZE = 1_000
 
-    def connection
-      @pipeline || @redis
-    end
-
     def each_key_batch_matching(key_pattern, &block)
       cursor = "0"
       begin

lib/kredis/types/proxy.rb

@@ -2,7 +2,7 @@ class Kredis::Types::Proxy
   require_relative "proxy/failsafe"
   include Failsafe
 
-  attr_accessor :redis, :key, :pipeline
+  attr_accessor :redis, :key
 
   def initialize(redis, key, **options)
     @redis, @key = redis, key
@@ -11,16 +11,17 @@ def initialize(redis, key, **options)
 
   def multi(*args, **kwargs, &block)
     redis.multi(*args, **kwargs) do |pipeline|
-      self.pipeline = pipeline
+      Thread.current[:pipeline] = pipeline
       block.call
-      self.pipeline = nil
+    ensure
+      Thread.current[:pipeline] = nil
     end
   end
 
   def method_missing(method, *args, **kwargs)
     Kredis.instrument :proxy, **log_message(method, *args, **kwargs) do
       failsafe do
-        (pipeline || redis).public_send method, key, *args, **kwargs
+        (Thread.current[:pipeline] || redis).public_send method, key, *args, **kwargs
       end
     end
   end