Redis tests

README

Redis tests

dispatcher.rb

#! /usr/bin/env ruby

require 'redis'
require 'json'

redis = Redis.new

lua_join_lists = "while redis.call('RPOPLPUSH', KEYS[1], KEYS[2]) do\nend\n"
redis.eval(lua_join_lists, ["events:dispatching", "events:queue"])

loop do
  while redis.exists("events:queue") do
    redis.rename("events:queue", "events:dispatching")

    redis.lrange("events:dispatching", 0, -1).each do |data|
      event = JSON.parse(data).first
      operator_id = event % 11
      redis.multi do
        redis.lpop("events:dispatching")
        redis.rpush("operator:#{operator_id}:events", data)
        redis.rpush("operators:queue", operator_id.to_s)
        redis.sadd("operators:known", operator_id.to_s)
      end
      puts "pushed event #{event} to operator:#{operator_id}:events"
      Kernel.sleep(0.05)
    end
  end

  Kernel.sleep(10)
end

Gemfile

gem 'redis'
gem 'json'

Gemfile.lock

GEM
  specs:
    json (1.8.3)
    redis (3.2.2)

PLATFORMS
  ruby

DEPENDENCIES
  json
  redis

BUNDLED WITH
   1.11.2

listener.rb

#! /usr/bin/env ruby

require 'redis'
require 'json'

redis = Redis.new

loop do
  event = rand(1000)
  redis.rpush('events:queue', JSON.dump([event]))
  puts "pushed event #{event}"

  Kernel.sleep(0.1)
end

monitor.rb

#! /usr/bin/env ruby

require 'redis'

redis = Redis.new

loop do
  sizes = (0..10).map { |x| redis.llen("operator:#{x}:events") }
  puts ("%4d" * 11 + " | %4d") % (sizes + [sizes.inject(0) { |s, v| s + v }])
  Kernel.sleep(1)
end

worker.rb

#! /usr/bin/env ruby

require 'redis'
require 'json'

redis = Redis.new

loop do
  _, operator_id = redis.blpop("operators:queue", 0)
  unless redis.sadd("operators:locked", operator_id)
    puts "Cannot lock operator #{operator_id}"
    next
  end

  puts "Locked operator #{operator_id}"

  key = "operator:#{operator_id}:events"
  while data = redis.lindex(key, 0) do
    event = JSON.parse(data).first
    puts "  #{operator_id}: #{event}"
    Kernel.sleep(0.1 + rand * 0.6)
    redis.lpop(key)
  end

  redis.srem("operators:locked", operator_id)

  redis.watch(key)
  if redis.llen(key) > 0
    # not empty
    redis.unwatch
  else
    redis.multi { redis.srem("operators:known", operator_id) }
  end

  puts "Unlocked operator #{operator_id}"
end