gistfile1.rb

require 'rubygems'
require 'eventmachine'
require 'em/buftok'
require 'base64'

class TwitterRealtimeJSON < EventMachine::Connection
  MAX_LINE_LENGTH = 16*1024

  attr_accessor :code
  attr_accessor :headers
  
  def initialize
    @code    = 0
    @headers = []
    @state   = :init
    @buffer  = BufferedTokenizer.new("\r", MAX_LINE_LENGTH)
  end
  
  def on_item &block
    @on_item_callback   = block
  end
  
  def stop
    close_connection
  end
  
  def receive_data data
    begin
      @buffer.extract(data).each do |line| 
        receive_line(line)
      end
    rescue Exception
      receive_error('overlength line')
      close_connection
      return
    end
  end
      
  def post_init
    data = []
    data << "GET /1/statuses/filter.json?track=TRACK_WHAT HTTP/1.1"
    data << "Host: stream.twitter.com"
    data << "Authorization: " + Base64.encode64('USER:PASSWORD')
    data << "\r\n"
    send_data data.join("\r\n") 
  end

  def receive_line ln
    case @state
    when :init
      parse_response_line ln
    when :headers
      parse_header_line ln
    when :stream
      parse_stream_line ln
    end
  end
  
  def receive_error e
    puts e
  end
  
  def parse_stream_line ln
    ln.strip!
    unless ln.empty?
      if ln[0,7] == '{"text"'
        @on_item_callback.call(ln) 
      end
    end
  end
  
  def parse_header_line ln
    ln.strip!
    if ln.empty?
      @state = :stream
    else
      headers << ln
    end
  end

  def parse_response_line ln
    if ln =~ /\AHTTP\/1\.[01] ([\d]{3})/
      @code = $1.to_i
      @state = :headers
    else
      receive_error('invalid response')
      close_connection
    end
  end

  def unbind
    puts "A connection has terminated"
  end
  
end

EventMachine::run {
  handler = EventMachine.connect "stream.twitter.com", 80, TwitterRealtimeJSON
  handler.on_item do |ln|
    print 'found: ' + ln
  end

  trap('TERM') {  
    handler.stop
    EventMachine.stop if EventMachine.reactor_running? 
  }
}
puts "The event loop has ended"