client-libraries/ruby_2/rubyredis.rb

   1 # RubyRedis is an alternative implementatin of Ruby client library written
   2 # by Salvatore Sanfilippo.
   3 #
   4 # The aim of this library is to create an alternative client library that is
   5 # much simpler and does not implement every command explicitly but uses
   6 # method_missing instead.
   7
   8 require 'socket'
   9 require 'set'
  10
  11 begin
  12     if (RUBY_VERSION >= '1.9')
  13         require 'timeout'
  14         RedisTimer = Timeout
  15     else
  16         require 'system_timer'
  17         RedisTimer = SystemTimer
  18     end
  19 rescue LoadError
  20     RedisTimer = nil
  21 end
  22
  23 class RedisClient
  24     BulkCommands = {
  25         "set"=>true, "setnx"=>true, "rpush"=>true, "lpush"=>true, "lset"=>true,
  26         "lrem"=>true, "sadd"=>true, "srem"=>true, "sismember"=>true,
  27         "echo"=>true, "getset"=>true, "smove"=>true
  28     }
  29
  30     ConvertToBool = lambda{|r| r == 0 ? false : r}
  31
  32     ReplyProcessor = {
  33         "exists" => ConvertToBool,
  34         "sismember"=> ConvertToBool,
  35         "sadd"=> ConvertToBool,
  36         "srem"=> ConvertToBool,
  37         "smove"=> ConvertToBool,
  38         "move"=> ConvertToBool,
  39         "setnx"=> ConvertToBool,
  40         "del"=> ConvertToBool,
  41         "renamenx"=> ConvertToBool,
  42         "expire"=> ConvertToBool,
  43         "keys" => lambda{|r| r.split(" ")},
  44         "info" => lambda{|r|
  45             info = {}
  46             r.each_line {|kv|
  47                 k,v = kv.split(":",2).map{|x| x.chomp}
  48                 info[k.to_sym] = v
  49             }
  50             info
  51         }
  52     }
  53
  54     Aliases = {
  55         "flush_db" => "flushdb",
  56         "flush_all" => "flushall",
  57         "last_save" => "lastsave",
  58         "key?" => "exists",
  59         "delete" => "del",
  60         "randkey" => "randomkey",
  61         "list_length" => "llen",
  62         "push_tail" => "rpush",
  63         "push_head" => "lpush",
  64         "pop_tail" => "rpop",
  65         "pop_head" => "lpop",
  66         "list_set" => "lset",
  67         "list_range" => "lrange",
  68         "list_trim" => "ltrim",
  69         "list_index" => "lindex",
  70         "list_rm" => "lrem",
  71         "set_add" => "sadd",
  72         "set_delete" => "srem",
  73         "set_count" => "scard",
  74         "set_member?" => "sismember",
  75         "set_members" => "smembers",
  76         "set_intersect" => "sinter",
  77         "set_intersect_store" => "sinterstore",
  78         "set_inter_store" => "sinterstore",
  79         "set_union" => "sunion",
  80         "set_union_store" => "sunionstore",
  81         "set_diff" => "sdiff",
  82         "set_diff_store" => "sdiffstore",
  83         "set_move" => "smove",
  84         "set_unless_exists" => "setnx",
  85         "rename_unless_exists" => "renamenx"
  86     }
  87
  88     def initialize(opts={})
  89         @host = opts[:host] || '127.0.0.1'
  90         @port = opts[:port] || 6379
  91         @db = opts[:db] || 0
  92         @timeout = opts[:timeout] || 0
  93         connect_to_server
  94     end
  95
  96     def to_s
  97         "Redis Client connected to #{@host}:#{@port} against DB #{@db}"
  98     end
  99
 100     def connect_to_server
 101         @sock = connect_to(@host,@port,@timeout == 0 ? nil : @timeout)
 102         call_command(["select",@db]) if @db != 0
 103     end
 104
 105     def connect_to(host, port, timeout=nil)
 106         # We support connect() timeout only if system_timer is availabe
 107         # or if we are running against Ruby >= 1.9
 108         # Timeout reading from the socket instead will be supported anyway.
 109         if @timeout != 0 and RedisTimer
 110             begin
 111                 sock = TCPSocket.new(host, port, 0)
 112             rescue Timeout::Error
 113                 @sock = nil
 114                 raise Timeout::Error, "Timeout connecting to the server"
 115             end
 116         else
 117             sock = TCPSocket.new(host, port, 0)
 118         end
 119
 120         # If the timeout is set we set the low level socket options in order
 121         # to make sure a blocking read will return after the specified number
 122         # of seconds. This hack is from memcached ruby client.
 123         if timeout
 124             secs = Integer(timeout)
 125             usecs = Integer((timeout - secs) * 1_000_000)
 126             optval = [secs, usecs].pack("l_2")
 127             sock.setsockopt Socket::SOL_SOCKET, Socket::SO_RCVTIMEO, optval
 128             sock.setsockopt Socket::SOL_SOCKET, Socket::SO_SNDTIMEO, optval
 129         end
 130         sock
 131     end
 132
 133     def method_missing(*argv)
 134         call_command(argv)
 135     end
 136
 137     def call_command(argv)
 138         # this wrapper to raw_call_command handle reconnection on socket
 139         # error. We try to reconnect just one time, otherwise let the error
 140         # araise.
 141         connect_to_server if !@sock
 142         begin
 143             raw_call_command(argv)
 144         rescue Errno::ECONNRESET
 145             @sock.close
 146             connect_to_server
 147             raw_call_command(argv)
 148         end
 149     end
 150
 151     def raw_call_command(argv)
 152         bulk = nil
 153         argv[0] = argv[0].to_s.downcase
 154         argv[0] = Aliases[argv[0]] if Aliases[argv[0]]
 155         if BulkCommands[argv[0]] and argv.length > 1
 156             bulk = argv[-1].to_s
 157             argv[-1] = bulk.length
 158         end
 159         @sock.write(argv.join(" ")+"\r\n")
 160         @sock.write(bulk+"\r\n") if bulk
 161
 162         # Post process the reply if needed
 163         processor = ReplyProcessor[argv[0]]
 164         processor ? processor.call(read_reply) : read_reply
 165     end
 166
 167     def select(*args)
 168         raise "SELECT not allowed, use the :db option when creating the object"
 169     end
 170
 171     def [](key)
 172         get(key)
 173     end
 174
 175     def []=(key,value)
 176         set(key,value)
 177     end
 178
 179     def sort(key, opts={})
 180         cmd = []
 181         cmd << "SORT #{key}"
 182         cmd << "BY #{opts[:by]}" if opts[:by]
 183         cmd << "GET #{[opts[:get]].flatten * ' GET '}" if opts[:get]
 184         cmd << "#{opts[:order]}" if opts[:order]
 185         cmd << "LIMIT #{opts[:limit].join(' ')}" if opts[:limit]
 186         call_command(cmd)
 187     end
 188
 189     def incr(key,increment=nil)
 190         call_command(increment ? ["incrby",key,increment] :  ["incr",key])
 191     end
 192
 193     def decr(key,decrement=nil)
 194         call_command(decrement ? ["decrby",key,decrement] :  ["decr",key])
 195     end
 196
 197     def read_reply
 198         # We read the first byte using read() mainly because gets() is
 199         # immune to raw socket timeouts.
 200         begin
 201             rtype = @sock.read(1)
 202         rescue Errno::EAGAIN
 203             # We want to make sure it reconnects on the next command after the
 204             # timeout. Otherwise the server may reply in the meantime leaving
 205             # the protocol in a desync status.
 206             @sock = nil
 207             raise Errno::EAGAIN, "Timeout reading from the socket"
 208         end
 209
 210         raise Errno::ECONNRESET,"Connection lost" if !rtype
 211         line = @sock.gets
 212         case rtype
 213         when "-"
 214             raise "-"+line.strip
 215         when "+"
 216             line.strip
 217         when ":"
 218             line.to_i
 219         when "$"
 220             bulklen = line.to_i
 221             return nil if bulklen == -1
 222             data = @sock.read(bulklen)
 223             @sock.read(2) # CRLF
 224             data
 225         when "*"
 226             objects = line.to_i
 227             return nil if bulklen == -1
 228             res = []
 229             objects.times {
 230                 res << read_reply
 231             }
 232             res
 233         else
 234             raise "Protocol error, got '#{rtype}' as initial reply byte"
 235         end
 236     end
 237 end