twitter_streams.rb

require 'oauth'
require 'oauth/client/em_http'
require 'yajl'
require 'pp'

User = Struct.new(:id, :token, :secret)

class Stream
  CONSUMER_KEY = 'xxxxxxxxxxxx'
  CONSUMER_SECRET = 'xxxxxxxxxxxxx'

  attr_reader :consumer, :access_token, :parser

  def initialize(u_token, u_secret)
    @consumer = OAuth::Consumer.new(CONSUMER_KEY, CONSUMER_SECRET, :site => "http://twitter.com")
    @access_token = OAuth::AccessToken.new @consumer, u_token, u_secret
    @parser = Yajl::Parser.new :symbolize_keys => true
    @parser.on_parse_complete = method(:dispatch)
  end

  def url
    raise 'not implemented'
  end

  def dispatch(data)
    pp data
  end

  def connect!
    request = EM::HttpRequest.new(url)
    @http = request.get(:timeout => 0) do |client|
      consumer.sign! client, access_token
    end

    @http.stream { |chunk|
      begin
        parser << chunk
      rescue Yajl::ParseError => e
        parse_error e
      end
    }
  end

  def disconnect!
    return unless @http
    @http.disconnect
    @http = nil
  end

  def reconnect!
    disconnect!
    EM.next_tick { connect! }
  end

  def parse_error(err)
    pp [:err, err]
  end
end

class SiteStream < Stream
  TOKEN = 'xxxxxxx'
  SECRET = 'xxxxxxxx'

  def initialize
    super TOKEN, SECRET
  end

  def track(users)
    @users = users
    reconnect!
  end

  def url
    URI.parse("http://betastream.twitter.com/2b/site.json?follow=#{@users.join(',')}")
  end
end

class UserStream < Stream
  attr_reader :user

  def self.track(user)
    s = self.new(user)
    s.connect!
    s
  end

  def initialize(user)
    super user.token, user.secret
  end

  def url
    URI.parse("https://userstream.twitter.com/2/user.json")
  end
end