class Redis

def [](key)
  self.get(key)
end

def []=(key,value)
  set(key,value)
end

def call_command(argv)
  @logger.debug { argv.inspect } if @logger
  # this wrapper to raw_call_command handle reconnection on socket
  # error. We try to reconnect just one time, otherwise let the error
  # araise.
  connect_to_server if !@sock
  begin
    raw_call_command(argv.dup)
  rescue Errno::ECONNRESET, Errno::EPIPE, Errno::ECONNABORTED
    @sock.close rescue nil
    @sock = nil
    connect_to_server
    raw_call_command(argv.dup)
  end
end

def connect_to(host, port, timeout=nil)
  # We support connect() timeout only if system_timer is availabe
  # or if we are running against Ruby >= 1.9
  # Timeout reading from the socket instead will be supported anyway.
  if @timeout != 0 and RedisTimer
    begin
      sock = TCPSocket.new(host, port)
    rescue Timeout::Error
      @sock = nil
      raise Timeout::Error, "Timeout connecting to the server"
    end
  else
    sock = TCPSocket.new(host, port)
  end
  sock.setsockopt Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1
  # If the timeout is set we set the low level socket options in order
  # to make sure a blocking read will return after the specified number
  # of seconds. This hack is from memcached ruby client.
  if timeout
    secs   = Integer(timeout)
    usecs  = Integer((timeout - secs) * 1_000_000)
    optval = [secs, usecs].pack("l_2")
    begin
      sock.setsockopt Socket::SOL_SOCKET, Socket::SO_RCVTIMEO, optval
      sock.setsockopt Socket::SOL_SOCKET, Socket::SO_SNDTIMEO, optval
    rescue Exception => ex
      # Solaris, for one, does not like/support socket timeouts.
      @logger.info "Unable to use raw socket timeouts: #{ex.class.name}: #{ex.message}" if @logger
    end
  end
  sock
end

def connect_to_server
  @sock = connect_to(@host, @port, @timeout == 0 ? nil : @timeout)
  call_command(["auth",@password]) if @password
  call_command(["select",@db]) unless @db == 0
end

def decr(key,decrement = nil)
  call_command(decrement ? ["decrby",key,decrement] : ["decr",key])
end

def get_size(string)
  string.respond_to?(:bytesize) ? string.bytesize : string.size
end

def incr(key, increment = nil)
  call_command(increment ? ["incrby",key,increment] : ["incr",key])
end

def initialize(options = {})
  @host    =  options[:host]    || '127.0.0.1'
  @port    = (options[:port]    || 6379).to_i
  @db      = (options[:db]      || 0).to_i
  @timeout = (options[:timeout] || 5).to_i
  @password = options[:password]
  @logger  =  options[:logger]
  @thread_safe = options[:thread_safe]
  @mutex = Mutex.new if @thread_safe
  @sock = nil
  @logger.info { self.to_s } if @logger
end

def mapped_mget(*keys)
  result = {}
  mget(*keys).each do |value|
    key = keys.shift
    result.merge!(key => value) unless value.nil?
  end
  result
end

def maybe_lock(&block)
  if @thread_safe
    @mutex.synchronize &block
  else
    block.call
  end
end

def method_missing(*argv)
  call_command(argv)
end

def pipelined(&block)
  pipeline = Pipeline.new self
  yield pipeline
  pipeline.execute
end

def process_command(command, argvv)
  @sock.write(command)
  argvv.map do |argv|
    processor = REPLY_PROCESSOR[argv[0]]
    processor ? processor.call(read_reply) : read_reply
  end
end

def quit
  call_command(['quit'])
rescue Errno::ECONNRESET
end

def raw_call_command(argvp)
  pipeline = argvp[0].is_a?(Array)
  unless pipeline
    argvv = [argvp]
  else
    argvv = argvp
  end
  if MULTI_BULK_COMMANDS[argvv.flatten[0].to_s]
    # TODO improve this code
    argvp   = argvv.flatten
    values  = argvp.pop.to_a.flatten
    argvp   = values.unshift(argvp[0])
    command = ["*#{argvp.size}"]
    argvp.each do |v|
      v = v.to_s
      command << "$#{get_size(v)}"
      command << v
    end
    command = command.map {|cmd| "#{cmd}\r\n"}.join
  else
    command = ""
    argvv.each do |argv|
      bulk = nil
      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]]
      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
  end
  results = maybe_lock { process_command(command, argvv) }
  
  return pipeline ? results : results[0]
end

def read_reply
  # We read the first byte using read() mainly because gets() is
  # immune to raw socket timeouts.
  begin
    rtype = @sock.read(1)
  rescue Errno::EAGAIN
    # We want to make sure it reconnects on the next command after the
    # timeout. Otherwise the server may reply in the meantime leaving
    # the protocol in a desync status.
    @sock = nil
    raise Errno::EAGAIN, "Timeout reading from the socket"
  end
  raise Errno::ECONNRESET,"Connection lost" if !rtype
  line = @sock.gets
  case rtype
  when MINUS
    raise MINUS + line.strip
  when PLUS
    line.strip
  when COLON
    line.to_i
  when DOLLAR
    bulklen = line.to_i
    return nil if bulklen == -1
    data = @sock.read(bulklen)
    @sock.read(2) # CRLF
    data
  when ASTERISK
    objects = line.to_i
    return nil if bulklen == -1
    res = []
    objects.times {
      res << read_reply
    }
    res
  else
    raise "Protocol error, got '#{rtype}' as initial reply byte"
  end
end

def select(*args)
  raise "SELECT not allowed, use the :db option when creating the object"
end

def server
  "#{@host}:#{@port}"
end

def set(key, value, expiry=nil)
  s = call_command([:set, key, value]) == OK
  expire(key, expiry) if s && expiry
  s
end

def sort(key, options = {})
  cmd = ["SORT"]
  cmd << key
  cmd << "BY #{options[:by]}" if options[:by]
  cmd << "GET #{[options[:get]].flatten * ' GET '}" if options[:get]
  cmd << "#{options[:order]}" if options[:order]
  cmd << "LIMIT #{options[:limit].join(' ')}" if options[:limit]
  call_command(cmd)
end

def to_s
  "Redis Client connected to #{server} against DB #{@db}"
end

def type(key)
  call_command(['type', key])
end