streamer.rb

require 'active_support'
require 'curb'
require 'yajl'

class TweetStreamer
  def initialize(username, password, host = 'stream.twitter.com')
    @username, @password = username, password
    @host = host
    @api_version = 1
  end
  
  class HttpError < RuntimeError; end
  
  def filter(params, &block)
    params.assert_valid_keys(:track, :follow)
    start(url('statuses/filter', params), &block)
  end
  
  def track(*keywords, &block)
    filter(:track => keywords.join(','), &block)
  end
  
  def sample(&block)
    start(url('statuses/sample'), &block)
  end
  
  private
  
  def start(url)
    parser = Yajl::Parser.new(:symbolize_keys => true)
    parser.on_parse_complete = proc{ |tweet| yield tweet }
    
    connection = Curl::Easy.new(url)
    connection.userpwd = "#{@username}:#{@password}"
    connection.on_body { |data| parser << data; data.length }
    connection.perform

    if connection.response_code.to_i > 200
      raise HttpError.new("Expected 200, got #{connection.response_code}")
    end
  end

  def url(api_path, params={})
    "http://#{@host}/#{@api_version}/#{api_path}.json?#{params.to_query}"
  end
end