Object-oriented Redis transaction

redis-tx.rb

require 'redis'
require 'set'

class RedisTransaction
  def initialize(redis)
    @redis = redis
    @watch_keys = Set.new
    @preconditions = []
    @executions = []
  end

  def watch(*keys)
    keys.each do |key|
      @watch_keys.add(key)
    end
  end

  def precondition(&block)
    @preconditions.push(block)
  end

  def execute(&block)
    @executions.push(block)
  end

  def commit
    @redis.watch(@watch_keys.to_a)
    unless check_preconditions
      @redis.unwatch
      raise 'Preconditions failed'
    end

    result = execute_all
    raise 'Commit aborted at EXEC' if result.nil?
    result
  end

  private

  def check_preconditions
    @preconditions.all? do |block|
      block.call(@redis)
    end
  end

  def execute_all
    @redis.multi do |multi|
      # Pipeline operations to Redis
      @executions.each do |block|
        block.call(multi)
      end
    end
  end
end

$redis = Redis.new
$redis.flushdb

tx = RedisTransaction.new($redis)

tx.watch('stream:0', 'stream:1')

tx.precondition do |redis|
  redis.llen('stream:0') == 0
end

tx.precondition do |redis|
  redis.llen('stream:1') == 0
end

tx.execute do |redis|
  redis.rpush('stream:1', ['event1', 'event2'])
end

tx.execute do |redis|
  redis.rpush('stream:0', ['event1'])
end