Module: EventMachine::Protocols::Redis
- Includes:
- EM::Deferrable
- Defined in:
- lib/em-redis/redis_protocol.rb
Defined Under Namespace
Classes: ParserError, ProtocolError, RedisError
Constant Summary collapse
- OK =
constants
"OK".freeze
- MINUS =
"-".freeze
- PLUS =
"+".freeze
- COLON =
":".freeze
- DOLLAR =
"$".freeze
- ASTERISK =
"*".freeze
- DELIM =
"\r\n".freeze
- BOOLEAN_PROCESSOR =
lambda{|r| r == 1 }
- REPLY_PROCESSOR =
{ "exists" => BOOLEAN_PROCESSOR, "sismember" => BOOLEAN_PROCESSOR, "sadd" => BOOLEAN_PROCESSOR, "srem" => BOOLEAN_PROCESSOR, "smove" => BOOLEAN_PROCESSOR, "zadd" => BOOLEAN_PROCESSOR, "zrem" => BOOLEAN_PROCESSOR, "move" => BOOLEAN_PROCESSOR, "setnx" => BOOLEAN_PROCESSOR, "del" => BOOLEAN_PROCESSOR, "renamenx" => BOOLEAN_PROCESSOR, "expire" => BOOLEAN_PROCESSOR, "select" => BOOLEAN_PROCESSOR, # not in redis gem "hset" => BOOLEAN_PROCESSOR, "hdel" => BOOLEAN_PROCESSOR, "hexists" => BOOLEAN_PROCESSOR, "keys" => lambda {|r| if r.is_a?(Array) r else r.split(" ") end }, "info" => lambda{|r| info = {} r.each_line {|kv| k,v = kv.split(":",2).map{|x| x.chomp} info[k.to_sym] = v } info }, "hgetall" => lambda{|r| Hash[*r] } }
- ALIASES =
{ "flush_db" => "flushdb", "flush_all" => "flushall", "last_save" => "lastsave", "key?" => "exists", "delete" => "del", "randkey" => "randomkey", "list_length" => "llen", "push_tail" => "rpush", "push_head" => "lpush", "pop_tail" => "rpop", "pop_head" => "lpop", "list_set" => "lset", "list_range" => "lrange", "list_trim" => "ltrim", "list_index" => "lindex", "list_rm" => "lrem", "set_add" => "sadd", "set_delete" => "srem", "set_count" => "scard", "set_member?" => "sismember", "set_members" => "smembers", "set_intersect" => "sinter", "set_intersect_store" => "sinterstore", "set_inter_store" => "sinterstore", "set_union" => "sunion", "set_union_store" => "sunionstore", "set_diff" => "sdiff", "set_diff_store" => "sdiffstore", "set_move" => "smove", "set_unless_exists" => "setnx", "rename_unless_exists" => "renamenx", "type?" => "type", "zset_add" => "zadd", "zset_count" => "zcard", "zset_range_by_score" => "zrangebyscore", "zset_reverse_range" => "zrevrange", "zset_range" => "zrange", "zset_delete" => "zrem", "zset_score" => "zscore", "zset_incr_by" => "zincrby", "zset_increment_by" => "zincrby", # these aliases aren't in redis gem "background_save" => 'bgsave', "async_save" => 'bgsave', "members" => 'smembers', "decrement_by" => "decrby", "decrement" => "decr", "increment_by" => "incrby", "increment" => "incr", "set_if_nil" => "setnx", "multi_get" => "mget", "random_key" => "randomkey", "random" => "randomkey", "rename_if_nil" => "renamenx", "tail_pop" => "rpop", "pop" => "rpop", "head_pop" => "lpop", "shift" => "lpop", "list_remove" => "lrem", "index" => "lindex", "trim" => "ltrim", "list_range" => "lrange", "range" => "lrange", "list_len" => "llen", "len" => "llen", "head_push" => "lpush", "unshift" => "lpush", "tail_push" => "rpush", "push" => "rpush", "add" => "sadd", "set_remove" => "srem", "set_size" => "scard", "member?" => "sismember", "intersect" => "sinter", "intersect_and_store" => "sinterstore", "members" => "smembers", "exists?" => "exists" }
- DISABLED_COMMANDS =
{ "monitor" => true, "sync" => true }
Class Method Summary collapse
-
.connect(*args) ⇒ Object
em hooks.
Instance Method Summary collapse
- #[]=(key, value) ⇒ Object
- #auth(password, &blk) ⇒ Object
- #call_command(argv, &blk) ⇒ Object
- #connection_completed ⇒ Object
- #decr(key, decrement = nil, &blk) ⇒ Object
- #dispatch_response(value) ⇒ Object
- #errback(&blk) ⇒ Object (also: #on_error)
- #exec(&blk) ⇒ Object
- #incr(key, increment = nil, &blk) ⇒ Object
- #initialize(options = {}) ⇒ Object
-
#mapped_mget(*keys) ⇒ Object
Similar to memcache.rb’s #get_multi, returns a hash mapping keys to values.
- #maybe_lock(&blk) ⇒ Object
- #method_missing(*argv, &blk) ⇒ Object
- #mset(*args, &blk) ⇒ Object
- #msetnx(*args, &blk) ⇒ Object
-
#multi ⇒ Object
I’m not sure autocommit is a good idea.
- #process_cmd(line) ⇒ Object
- #quit(&blk) ⇒ Object
-
#receive_data(data) ⇒ Object
19Feb09 Switched to a custom parser, LineText2 is recursive and can cause stack overflows when there is too much data.
- #select(db, &blk) ⇒ Object
- #set(key, value, expiry = nil) ⇒ Object
- #sort(key, options = {}, &blk) ⇒ Object
-
#type(key, &blk) ⇒ Object
Ruby defines a now deprecated type method so we need to override it here since it will never hit method_missing.
- #unbind ⇒ Object
Dynamic Method Handling
This class handles dynamic methods through the method_missing method
#method_missing(*argv, &blk) ⇒ Object
251 252 253 |
# File 'lib/em-redis/redis_protocol.rb', line 251 def method_missing(*argv, &blk) call_command(argv, &blk) end |
Class Method Details
.connect(*args) ⇒ Object
em hooks
304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 |
# File 'lib/em-redis/redis_protocol.rb', line 304 def self.connect(*args) case args.length when 0 = {} when 1 arg = args.shift case arg when Hash then = arg when String then = {:host => arg} else raise ArgumentError, 'first argument must be Hash or String' end when 2 = {:host => args[0], :port => args[1]} else raise ArgumentError, "wrong number of arguments (#{args.length} for 1)" end [:host] ||= '127.0.0.1' [:port] = ([:port] || 6379).to_i EM.connect [:host], [:port], self, end |
Instance Method Details
#[]=(key, value) ⇒ Object
145 146 147 |
# File 'lib/em-redis/redis_protocol.rb', line 145 def []=(key,value) set(key,value) end |
#auth(password, &blk) ⇒ Object
179 180 181 182 |
# File 'lib/em-redis/redis_protocol.rb', line 179 def auth(password, &blk) @password = password call_command(['auth', password], &blk) end |
#call_command(argv, &blk) ⇒ Object
265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 |
# File 'lib/em-redis/redis_protocol.rb', line 265 def call_command(argv, &blk) argv = argv.dup argv[0] = argv[0].to_s.downcase argv[0] = ALIASES[argv[0]] if ALIASES[argv[0]] raise "#{argv[0]} command is disabled" if DISABLED_COMMANDS[argv[0]] command = "" command << "*#{argv.size}\r\n" argv.each do |a| a = a.to_s command << "$#{get_size(a)}\r\n" command << a command << "\r\n" end @logger.debug { "*** sending: #{command}" } if @logger maybe_lock do @redis_callbacks << [REPLY_PROCESSOR[argv[0]], blk] send_data command end end |
#connection_completed ⇒ Object
347 348 349 350 351 352 353 354 355 356 357 |
# File 'lib/em-redis/redis_protocol.rb', line 347 def connection_completed @logger.debug { "Connected to #{@host}:#{@port}" } if @logger @redis_callbacks = [] @previous_multibulks = [] @multibulk_n = false @reconnecting = false @connected = true succeed end |
#decr(key, decrement = nil, &blk) ⇒ Object
170 171 172 |
# File 'lib/em-redis/redis_protocol.rb', line 170 def decr(key, decrement = nil, &blk) call_command(decrement ? ["decrby",key,decrement] : ["decr",key], &blk) end |
#dispatch_response(value) ⇒ Object
424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 |
# File 'lib/em-redis/redis_protocol.rb', line 424 def dispatch_response(value) if @multibulk_n @multibulk_values << value @multibulk_n -= 1 if @multibulk_n == 0 value = @multibulk_values @multibulk_n,@multibulk_values = @previous_multibulks.pop if @multibulk_n dispatch_response(value) return end else return end end processor, blk = @redis_callbacks.shift value = processor.call(value) if processor blk.call(value) if blk end |
#errback(&blk) ⇒ Object Also known as: on_error
246 247 248 |
# File 'lib/em-redis/redis_protocol.rb', line 246 def errback(&blk) @error_callback = blk end |
#exec(&blk) ⇒ Object
207 208 209 |
# File 'lib/em-redis/redis_protocol.rb', line 207 def exec(&blk) call_command(['exec'], &blk) end |
#incr(key, increment = nil, &blk) ⇒ Object
166 167 168 |
# File 'lib/em-redis/redis_protocol.rb', line 166 def incr(key, increment = nil, &blk) call_command(increment ? ["incrby",key,increment] : ["incr",key], &blk) end |
#initialize(options = {}) ⇒ Object
325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 |
# File 'lib/em-redis/redis_protocol.rb', line 325 def initialize( = {}) @host = [:host] @port = [:port] @db = ([:db] || 0).to_i @password = [:password] @logger = [:logger] @error_callback = lambda do |code| err = RedisError.new err.code = code raise err, "Redis server returned error code: #{code}" end # These commands should be first auth_and_select_db end |
#mapped_mget(*keys) ⇒ Object
Similar to memcache.rb’s #get_multi, returns a hash mapping keys to values.
186 187 188 189 190 191 192 193 194 195 |
# File 'lib/em-redis/redis_protocol.rb', line 186 def mapped_mget(*keys) mget(*keys) do |response| result = {} response.each do |value| key = keys.shift result.merge!(key => value) unless value.nil? end yield result if block_given? end end |
#maybe_lock(&blk) ⇒ Object
255 256 257 258 259 260 261 262 263 |
# File 'lib/em-redis/redis_protocol.rb', line 255 def maybe_lock(&blk) if !EM.reactor_thread? EM.schedule { maybe_lock(&blk) } elsif @connected yield else callback { yield } end end |
#mset(*args, &blk) ⇒ Object
228 229 230 231 232 233 234 235 |
# File 'lib/em-redis/redis_protocol.rb', line 228 def mset(*args, &blk) hsh = args.pop if Hash === args.last if hsh call_command(hsh.to_a.flatten.unshift(:mset), &blk) else call_command(args.unshift(:mset), &blk) end end |
#msetnx(*args, &blk) ⇒ Object
237 238 239 240 241 242 243 244 |
# File 'lib/em-redis/redis_protocol.rb', line 237 def msetnx(*args, &blk) hsh = args.pop if Hash === args.last if hsh call_command(hsh.to_a.flatten.unshift(:msetnx), &blk) else call_command(args.unshift(:msetnx), &blk) end end |
#multi ⇒ Object
I’m not sure autocommit is a good idea. For example:
r.multi { r.set('a', 'b') { raise "kaboom" } }
will commit “a” and will stop EM
215 216 217 218 219 220 221 222 223 224 225 226 |
# File 'lib/em-redis/redis_protocol.rb', line 215 def multi call_command(['multi']) if block_given? begin yield self exec rescue Exception => e discard raise e end end end |
#process_cmd(line) ⇒ Object
375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 |
# File 'lib/em-redis/redis_protocol.rb', line 375 def process_cmd(line) @logger.debug { "*** processing #{line}" } if @logger # first character of buffer will always be the response type reply_type = line[0, 1] reply_args = line.slice(1..-3) # remove type character and \r\n case reply_type #e.g. -MISSING when MINUS @redis_callbacks.shift # throw away the cb? @error_callback.call(reply_args) # e.g. +OK when PLUS dispatch_response(reply_args) # e.g. $3\r\nabc\r\n # 'bulk' is more complex because it could be part of multi-bulk when DOLLAR data_len = Integer(reply_args) if data_len == -1 # expect no data; return nil dispatch_response(nil) elsif @buffer.size >= data_len + 2 # buffer is full of expected data dispatch_response(@buffer.slice!(0, data_len)) @buffer.slice!(0,2) # tossing \r\n else # buffer isn't full or nil # TODO: don't control execution with exceptions raise ParserError end #e.g. :8 when COLON dispatch_response(Integer(reply_args)) #e.g. *2\r\n$1\r\na\r\n$1\r\nb\r\n when ASTERISK multibulk_count = Integer(reply_args) if multibulk_count == -1 || multibulk_count == 0 dispatch_response([]) else if @multibulk_n @previous_multibulks << [@multibulk_n, @multibulk_values] end @multibulk_n = multibulk_count @multibulk_values = [] end # Whu? else # TODO: get rid of this exception raise ProtocolError, "reply type not recognized: #{line.strip}" end end |
#quit(&blk) ⇒ Object
203 204 205 |
# File 'lib/em-redis/redis_protocol.rb', line 203 def quit(&blk) call_command(['quit'], &blk) end |
#receive_data(data) ⇒ Object
19Feb09 Switched to a custom parser, LineText2 is recursive and can cause
stack overflows when there is too much data.
include EM::P::LineText2
362 363 364 365 366 367 368 369 370 371 372 373 |
# File 'lib/em-redis/redis_protocol.rb', line 362 def receive_data(data) (@buffer ||= '') << data while index = @buffer.index(DELIM) begin line = @buffer.slice!(0, index+2) process_cmd line rescue ParserError @buffer[0...0] = line break end end end |
#select(db, &blk) ⇒ Object
174 175 176 177 |
# File 'lib/em-redis/redis_protocol.rb', line 174 def select(db, &blk) @db = db.to_i call_command(['select', @db], &blk) end |
#set(key, value, expiry = nil) ⇒ Object
149 150 151 152 153 154 |
# File 'lib/em-redis/redis_protocol.rb', line 149 def set(key, value, expiry=nil) call_command([:set, key, value]) do |s| yield s if block_given? end expire(key, expiry) if expiry end |
#sort(key, options = {}, &blk) ⇒ Object
156 157 158 159 160 161 162 163 164 |
# File 'lib/em-redis/redis_protocol.rb', line 156 def sort(key, ={}, &blk) cmd = ["SORT"] cmd << key cmd << ["BY", [:by]] if [:by] cmd << [[:get]].flatten.map { |key| ["GET", key] } if [:get] cmd << [:order].split(/\s+/) if [:order] cmd << ["LIMIT", [:limit]] if [:limit] call_command(cmd.flatten, &blk) end |
#type(key, &blk) ⇒ Object
Ruby defines a now deprecated type method so we need to override it here since it will never hit method_missing
199 200 201 |
# File 'lib/em-redis/redis_protocol.rb', line 199 def type(key, &blk) call_command(['type', key], &blk) end |
#unbind ⇒ Object
446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 |
# File 'lib/em-redis/redis_protocol.rb', line 446 def unbind @logger.debug { "Disconnected" } if @logger if @connected || @reconnecting EM.add_timer(1) do @logger.debug { "Reconnecting to #{@host}:#{@port}" } if @logger reconnect @host, @port auth_and_select_db end @connected = false @reconnecting = true @deferred_status = nil else # TODO: get rid of this exception raise 'Unable to connect to redis server' end end |