Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
tree: aa5ae6bd01
Fetching contributors…

Cannot retrieve contributors at this time

file 202 lines (171 sloc) 5.099 kb
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202
module EventMachine::Hiredis
  class Client
    PUBSUB_MESSAGES = %w{message pmessage}.freeze

    include EventMachine::Hiredis::EventEmitter
    include EM::Deferrable

    attr_reader :host, :port, :password, :db

    def self.connect(host = 'localhost', port = 6379)
      new(host, port).connect
    end

    def initialize(host, port, password = nil, db = nil)
      @host, @port, @password, @db = host, port, password, db
      @subs, @psubs, @defs = [], [], []
      @closing_connection = false
    end

    def connect
      @connection = EM.connect(@host, @port, Connection, @host, @port)

      @connection.on(:closed) do
        if @connected
          @defs.each { |d| d.fail("Redis disconnected") }
          @defs = []
          @deferred_status = nil
          @connected = false
          unless @closing_connection
            @reconnecting = true
            reconnect
          end
        else
          unless @closing_connection
            EM.add_timer(1) { reconnect }
          end
        end
      end

      @connection.on(:connected) do
        @connected = true

        auth(@password) if @password
        select(@db) if @db

        @subs.each { |s| method_missing(:subscribe, s) }
        @psubs.each { |s| method_missing(:psubscribe, s) }
        succeed

        if @reconnecting
          @reconnecting = false
          emit(:reconnected)
        end
      end

      @connection.on(:message) do |reply|
        if RuntimeError === reply
          raise "Replies out of sync: #{reply.inspect}" if @defs.empty?
          deferred = @defs.shift
          deferred.fail(reply) if deferred
        else
          if reply && PUBSUB_MESSAGES.include?(reply[0]) # reply can be nil
            kind, subscription, d1, d2 = *reply

            case kind.to_sym
            when :message
              emit(:message, subscription, d1)
            when :pmessage
              emit(:pmessage, subscription, d1, d2)
            end
          else
            if @defs.empty?
              if @monitoring
                emit(:monitor, reply)
              else
                raise "Replies out of sync: #{reply.inspect}"
              end
            else
              deferred = @defs.shift
              deferred.succeed(reply) if deferred
            end
          end
        end
      end

      @connected = false
      @reconnecting = false

      return self
    end

    # Indicates that commands have been sent to redis but a reply has not yet
    # been received
    #
    # This can be useful for example to avoid stopping the
    # eventmachine reactor while there are outstanding commands
    #
    def pending_commands?
      @connected && @defs.size > 0
    end

    def connected?
      @connected
    end

    def subscribe(channel)
      @subs << channel
      method_missing(:subscribe, channel)
    end

    def unsubscribe(channel)
      @subs.delete(channel)
      method_missing(:unsubscribe, channel)
    end

    def psubscribe(channel)
      @psubs << channel
      method_missing(:psubscribe, channel)
    end

    def punsubscribe(channel)
      @psubs.delete(channel)
      method_missing(:punsubscribe, channel)
    end

    def select(db, &blk)
      @db = db
      method_missing(:select, db, &blk)
    end

    def auth(password, &blk)
      @password = password
      method_missing(:auth, password, &blk)
    end

    def monitor(&blk)
      @monitoring = true
      method_missing(:monitor, &blk)
    end

    def info(&blk)
      hash_processor = lambda do |response|
        info = {}
        response.each_line do |line|
          key, value = line.chomp.split(":", 2)
          info[key.to_sym] = value if value && key =~ /^[^#]/
        end
        blk.call(info)
      end
      method_missing(:info, &hash_processor)
    end

    def info_commandstats(&blk)
      hash_processor = lambda do |response|
        commands = {}
        response.each_line do |line|
          command, data = line.split(':')
          if data
            c = commands[command.sub('cmdstat_', '').to_sym] = {}
            data.split(',').each do |d|
              k, v = d.split('=')
              c[k.to_sym] = v =~ /\./ ? v.to_f : v.to_i
            end
          end
        end
        blk.call(commands)
      end
      method_missing(:info, 'commandstats', &hash_processor)
    end

    def close_connection
      @closing_connection = true
      @connection.close_connection_after_writing
      @defs.each
    end

    private

    def method_missing(sym, *args)
      deferred = EM::DefaultDeferrable.new
      # Shortcut for defining the callback case with just a block
      deferred.callback { |result| yield(result) } if block_given?

      if @connected
        @connection.send_command(sym, *args)
        @defs.push(deferred)
      else
        callback do
          @connection.send_command(sym, *args)
          @defs.push(deferred)
        end
      end

      deferred
    end

    def reconnect
      EventMachine::Hiredis.logger.debug("Trying to reconnect to Redis")
      @connection.reconnect @host, @port
    end
  end
end
Something went wrong with that request. Please try again.