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
- BULK_COMMANDS =
{ "set" => true, "setnx" => true, "rpush" => true, "lpush" => true, "lset" => true, "lrem" => true, "sadd" => true, "srem" => true, "sismember" => true, "echo" => true, "getset" => true, "smove" => true, "zadd" => true, "zincrby" => true, "zrem" => true, "zscore" => true, "hget" => true, "hdel" => true, "hexists" => true }
- MULTI_BULK_COMMANDS =
{ "mset" => true, "msetnx" => true, "hset" => true, # these aliases aren't in redis gem "multi_get" => true }
- 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
281 282 283 |
# File 'lib/em-redis/redis_protocol.rb', line 281 def method_missing(*argv, &blk) call_command(argv, &blk) end |
Class Method Details
.connect(*args) ⇒ Object
em hooks
344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 |
# File 'lib/em-redis/redis_protocol.rb', line 344 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
175 176 177 |
# File 'lib/em-redis/redis_protocol.rb', line 175 def []=(key,value) set(key,value) end |
#auth(password, &blk) ⇒ Object
209 210 211 212 |
# File 'lib/em-redis/redis_protocol.rb', line 209 def auth(password, &blk) @password = password call_command(['auth', password], &blk) end |
#call_command(argv, &blk) ⇒ Object
295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 |
# File 'lib/em-redis/redis_protocol.rb', line 295 def call_command(argv, &blk) argv = argv.dup argv[0] = argv[0].to_s.downcase if MULTI_BULK_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 else command = "" bulk = nil argv[0] = ALIASES[argv[0]] if ALIASES[argv[0]] raise "#{argv[0]} command is disabled" if DISABLED_COMMANDS[argv[0]] if BULK_COMMANDS[argv[0]] and argv.length > 1 bulk = argv[-1].to_s argv[-1] = get_size(bulk) end command << "#{argv.join(' ')}\r\n" command << "#{bulk}\r\n" if bulk 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
387 388 389 390 391 392 393 394 395 396 397 |
# File 'lib/em-redis/redis_protocol.rb', line 387 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
200 201 202 |
# File 'lib/em-redis/redis_protocol.rb', line 200 def decr(key, decrement = nil, &blk) call_command(decrement ? ["decrby",key,decrement] : ["decr",key], &blk) end |
#dispatch_response(value) ⇒ Object
464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 |
# File 'lib/em-redis/redis_protocol.rb', line 464 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
276 277 278 |
# File 'lib/em-redis/redis_protocol.rb', line 276 def errback(&blk) @error_callback = blk end |
#exec(&blk) ⇒ Object
237 238 239 |
# File 'lib/em-redis/redis_protocol.rb', line 237 def exec(&blk) call_command(['exec'], &blk) end |
#incr(key, increment = nil, &blk) ⇒ Object
196 197 198 |
# File 'lib/em-redis/redis_protocol.rb', line 196 def incr(key, increment = nil, &blk) call_command(increment ? ["incrby",key,increment] : ["incr",key], &blk) end |
#initialize(options = {}) ⇒ Object
365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 |
# File 'lib/em-redis/redis_protocol.rb', line 365 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.
216 217 218 219 220 221 222 223 224 225 |
# File 'lib/em-redis/redis_protocol.rb', line 216 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
285 286 287 288 289 290 291 292 293 |
# File 'lib/em-redis/redis_protocol.rb', line 285 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
258 259 260 261 262 263 264 265 |
# File 'lib/em-redis/redis_protocol.rb', line 258 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
267 268 269 270 271 272 273 274 |
# File 'lib/em-redis/redis_protocol.rb', line 267 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
245 246 247 248 249 250 251 252 253 254 255 256 |
# File 'lib/em-redis/redis_protocol.rb', line 245 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
415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 |
# File 'lib/em-redis/redis_protocol.rb', line 415 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
233 234 235 |
# File 'lib/em-redis/redis_protocol.rb', line 233 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
402 403 404 405 406 407 408 409 410 411 412 413 |
# File 'lib/em-redis/redis_protocol.rb', line 402 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
204 205 206 207 |
# File 'lib/em-redis/redis_protocol.rb', line 204 def select(db, &blk) @db = db.to_i call_command(['select', @db], &blk) end |
#set(key, value, expiry = nil) ⇒ Object
179 180 181 182 183 184 |
# File 'lib/em-redis/redis_protocol.rb', line 179 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
186 187 188 189 190 191 192 193 194 |
# File 'lib/em-redis/redis_protocol.rb', line 186 def sort(key, ={}, &blk) cmd = ["SORT"] cmd << key cmd << "BY #{[:by]}" if [:by] cmd << "GET #{[[:get]].flatten * ' GET '}" if [:get] cmd << "#{[:order]}" if [:order] cmd << "LIMIT #{[:limit].join(' ')}" if [:limit] call_command(cmd, &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
229 230 231 |
# File 'lib/em-redis/redis_protocol.rb', line 229 def type(key, &blk) call_command(['type', key], &blk) end |
#unbind ⇒ Object
486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 |
# File 'lib/em-redis/redis_protocol.rb', line 486 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 |