Me learning eventmachine basics... a redis client.

redis-eventmachine-toy.rb

require 'rubygems'
require 'eventmachine'

class RedisClient < EventMachine::Connection

  include EM::Protocols::LineText2
  include EM::Deferrable
  
  unless defined? Rcrlf
    Rcrlf = "\r\n".freeze
  end
  
  def self.connect(host = 'localhost', port = 6379)
    EM.connect host, port, self, host, port
    self.set_comm_inactivity_timeout 2
  end

  def initialize(host = 'localhost', port = 6379)
    @host, @port = host, port
    @commands = []
    @bulk_cbs = []
  end

  def connection_completed
    @values = {}
    @reconnecting = false
    @connected = true
  end
  
  def info
    raise ArgumentError unless block_given?
    @commands << "INFO"
    send_data("INFO" + Rcrlf)
    @bulk_cbs << proc { |lines|
      yield lines
    }
  end

  def set(key, val)
    @commands << "SET "
    send_data("SET #{key} #{val.size}" + Rcrlf + val + Rcrlf)   
  end

  def get(key, &cb)
    raise ArgumentError unless block_given?
     @commands << "GET "
     send_data("GET #{key}" + Rcrlf)
     @bulk_cbs << cb 
  end

  def select_db(db_number)
    @commands << "SELECT "
    send_data("SELECT #{db_number}" + Rcrlf)    
  end
  
  def post_init
    #puts "post init called"
  end

  def on_bulk_reply
    cb = @bulk_cbs.shift
    @commands.shift
    
    if @bulk_reply_lines.size == 1
      cb.call(@bulk_reply_lines.first)
    else
      cb.call(@bulk_reply_lines)
    end
    
    @bulk_reply = false
  end
  
  def receive_line(line)

    if line == ""
      on_bulk_reply
      return
    end
    
    case line[0].chr
    when "+"
      puts "#{@commands.first} status code reply: #{line}"
      @commands.shift
    when "-"
      puts "#{@commands.first} error code reply: #{line}"
      @commands.shift
    when "$"
      puts "bulk reply"
      @bulk_reply = true
      @bulk_reply_length = line[1..-1].to_i
      @bulk_reply_lines = []
      puts "bulk reply of size #{@bulk_reply_length}"
    else
      if @bulk_reply
        @bulk_reply_lines << line    
        if @bulk_reply_lines.join.size == @bulk_reply_length
          on_bulk_reply
        end
      else
        puts "Got a line:[#{line}]"
      end
    end
end

 def unbind
   if @connected or @reconnecting
     EM.add_timer(1){ reconnect @host, @port }
     @connected = false
     @reconnecting = true
     @deferred_status = nil
   else
     raise 'Unable to connect to redis server'
   end
 end
end

EventMachine::run {
  redis = EventMachine::connect '127.0.0.1', 6379, RedisClient 

  redis.info { |info|
    p "Server Info:  #{info.inspect}"
  }
  
  redis.select_db 3
  
  redis.set('foo', 'foobar')
  
  redis.get('foo') {|value|
    p "Value for 'foo' is #{value}"
  }
}