class Redis::Pipeline
def call(command, timeout: nil, &block)
def call(command, timeout: nil, &block) # A pipeline that contains a shutdown should not raise ECONNRESET when # the connection is gone. @shutdown = true if command.first == :shutdown future = Future.new(command, block, timeout) @futures << future future end
def call_pipeline(pipeline)
def call_pipeline(pipeline) @shutdown = true if pipeline.shutdown? @futures.concat(pipeline.futures) @db = pipeline.db nil end
def call_with_timeout(command, timeout, &block)
def call_with_timeout(command, timeout, &block) call(command, timeout: timeout, &block) end
def commands
def commands @futures.map(&:_command) end
def empty?
def empty? @futures.empty? end
def finish(replies, &blk)
def finish(replies, &blk) if blk futures.each_with_index.map do |future, i| future._set(blk.call(replies[i])) end else futures.each_with_index.map do |future, i| future._set(replies[i]) end end end
def initialize(client)
def initialize(client) @client = client.is_a?(Pipeline) ? client.client : client @with_reconnect = true @shutdown = false @futures = [] end
def shutdown?
def shutdown? @shutdown end
def timeout
def timeout client.timeout end
def timeouts
def timeouts @futures.map(&:timeout) end
def with_reconnect(val = true)
def with_reconnect(val = true) @with_reconnect = false unless val yield end
def with_reconnect?
def with_reconnect? @with_reconnect end
def without_reconnect(&blk)
def without_reconnect(&blk) with_reconnect(false, &blk) end
def without_reconnect?
def without_reconnect? !@with_reconnect end