Class: EM::Redis::Client

Inherits:
Connection
  • Object
show all
Includes:
Deferrable
Defined in:
lib/em/redis.rb,
lib/em-synchrony/em-redislite.rb

Constant Summary collapse

DB_ERROR = 
'lost redis connection'
COMMANDS = 
%w(
  APPEND AUTH BLPOP BRPOP BRPOPLPUSH DECR DECRBY DEL DISCARD DUMP ECHO EXEC EXISTS EXPIRE EXPIREAT
  GET GETBIT GETRANGE GETSET HDEL HEXISTS HGET HGETALL HINCRBY HINCRBYFLOAT HKEYS HLEN HMGET HMSET
  HSET HSETNX HVALS INCR INCRBY INCRBYFLOAT KEYS LINDEX LINSERT LLEN LPOP LPUSH LPUSHX LRANGE LREM
  LSET LTRIM MGET MIGRATE MOVE MSET MSETNX MULTI OBJECT PERSIST PEXPIRE PEXPIREAT PING PSETEX PTTL
  QUIT RANDOMKEY RENAME RENAMENX RESTORE RPOP RPOPLPUSH RPUSH RPUSHX SADD SCARD SDIFF SDIFFSTORE
  SELECT SET SETBIT SETEX SETNX SETRANGE SINTER SINTERSTORE SISMEMBER SMEMBERS SMOVE SORT SPOP
  SRANDMEMBER SREM STRLEN SUNION SUNIONSTORE TTL TYPE UNWATCH WATCH ZADD ZCARD ZCOUNT ZINCRBY
  ZINTERSTORE ZRANGE ZRANGEBYSCORE ZRANK ZREM ZREMRANGEBYRANK ZREMRANGEBYSCORE ZREVRANGE
  ZREVRANGEBYSCORE ZREVRANK ZSCORE ZUNIONSTORE
).freeze

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Client

Returns a new instance of Client.



29
30
31
 
# File 'lib/em/redis.rb', line 29

def initialize options = {}
  @options = options
end

Instance Method Details

#error? ⇒ Boolean

Returns:

  • (Boolean) 


41
42
43
 
# File 'lib/em/redis.rb', line 41

def error?
  signature ? super : true
end

#on_error(msg, dns_error = false) ⇒ Object 



58
59
60
 
# File 'lib/em/redis.rb', line 58

def on_error msg, dns_error = false
  unbind(msg)
end

#post_init ⇒ Object 



33
34
35
36
37
38
39
 
# File 'lib/em/redis.rb', line 33

def post_init
  @buffer     = ''
  @want_bytes = 0
  @want_lines = 0
  @pool       = []
  @lines      = []
end

#process_bytes(data) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/em/redis.rb', line 93

def process_bytes data
  @want_bytes = 0
  if @want_lines > 0
    @lines.push(data)
    if @lines.length == @want_lines
      @pool.shift.succeed(@lines)
      @lines, @want_lines = [], 0
    end
  else
    @pool.shift.succeed(data)
  end
end

#process_response(data) ⇒ Object 



106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/em/redis.rb', line 106

def process_response data
  case data[0]
    when '+' then @pool.shift.succeed(data[1..-1])
    when '-' then @pool.shift.fail(Error.new(data[1..-1]))
    when ':' then @pool.shift.succeed(data[1..-1])
    when '$' then @want_bytes = data[1..-1].to_i
    when '*' then @want_lines = data[1..-1].to_i
    else     @pool.shift.fail DataError.new("Unknown data format: #{data}")
  end

  process_bytes(nil) if @want_bytes < 0
end

#receive_data(data) ⇒ Object 



80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/em/redis.rb', line 80

def receive_data data
  @buffer << data
  while index = @buffer.index("\r\n")
    if @want_bytes > 0
      break unless @buffer.bytesize >= @want_bytes + 2
      data = @buffer.slice!(0, @want_bytes + 2)
      process_bytes(data[0..@want_bytes-1])
    else
      process_response @buffer.slice!(0, index+2).strip
    end
  end
end

#reconnect! ⇒ Object 



66
67
68
 
# File 'lib/em/redis.rb', line 66

def reconnect!
  EM.reconnect @options[:host], @options[:port], self
end

#reset_errback(&block) ⇒ Object 



62
63
64
 
# File 'lib/em/redis.rb', line 62

def reset_errback &block
  @errbacks = [ block ]
end

#send_command(name, *args) ⇒ Object Also known as: command 



45
46
47
48
49
50
51
52
53
54
 
# File 'lib/em/redis.rb', line 45

def send_command name, *args
  defer = DefaultDeferrable.new
  return defer.tap {defer.fail(ConnectionError.new(DB_ERROR))} if error?

  @pool << defer
  send_data "*#{args.length + 1}\r\n" +
            "$#{name.length}\r\n#{name}\r\n" +
            args.map(&:to_s).inject('') {|a,v| a + "$#{v.bytesize}\r\n#{v}\r\n"}
  defer
end

#unbind(msg = DB_ERROR) ⇒ Object 



70
71
72
73
74
75
76
77
78
 
# File 'lib/em/redis.rb', line 70

def unbind msg = DB_ERROR
  close_connection
  error      = ConnectionError.new(msg)
  @signature = nil

  @pool.each {|r| r.fail(error) }
  @pool = []
  fail error
end