redis_subscribe.rb

require 'stringio'

class RedisSubscribe < EM::Connection
  def self.connect(host, port, pass = '')
    Rails.logger.debug host
    Rails.logger.debug port

    client = EM.connect host, port, self

    client.auth pass if pass.present?
  end

  def post_init
    call_command 'publish', 'clientconnected', 'rails'
    call_command 'subscribe', 'channelone'
  end

  def receive_data(data)
    buffer = StringIO.new(data)
    begin
      parts = read_response(buffer)
    end while !buffer.eof?
  end

  private

  def read_response(buffer)
    type = buffer.read(1)
    case type
    when ':'
      buffer.gets.to_i
    when '*'
      size = buffer.gets.to_i
      size.times.map { read_object(buffer) }
    else
      raise "unsupported response type: #{type}"
    end
  end

  def read_object(data)
    type = data.read(1)
    case type
    when ':' # integer
      data.gets.to_i
    when '$'
      size = data.gets
      str = data.read(size.to_i)
      data.read(2) # crlf
      str
    else
      raise "read for object of type #{type} not implemented"
    end
  end

  def call_command(*args)
    command = "*#{args.size}\r\n"
    args.each { |a|
      command << "$#{a.to_s.size}\r\n"
      command << a.to_s
      command << "\r\n"
    }
    send_data command
  end
end

Borrowed/Stole a bunch from https://gist.github.com/tobym/352068