simple TwitStream using EventMachine

gistfile1.rb

#!/usr/bin/env ruby

# usage;
# script/twitstream start
# script/twitstream stop
# script/twitstream restart
# pid and log at RAILS_ROOT/tmp/pids

# where config/twitstream.yml looks like this;

#  development:
#   path: /1/statuses/filter.json
#   auth: twitter_user:twitter_pass
#   method: POST
#   content: follow=83442970
#   # 83442970 is the twitter id follow stream to listen on

require 'rubygems'
require 'daemons'

RAILS_HOME       = File.expand_path(File.join(File.dirname(__FILE__),".."))
PID_HOME         = File.join(RAILS_HOME, 'tmp', 'pids')
ENV['RAILS_ENV'] = (ENV['RAILS_ENV'] || 'development')

# load config
begin
  @twitconfig = YAML::load(File.open("#{RAILS_HOME}/config/twitstream.yml"))["#{ENV['RAILS_ENV']}"]
rescue StandardError => e
  puts "\nOoops ! problems loading twitstream config, check config/twitstream.yml file for a '#{ENV['RAILS_ENV']}' environment: #{e}\n"
  exit 0
end                     

puts "TwitStream (#{ENV['RAILS_ENV']}): #{ARGV.first}\n"

# dameonized event loop
Daemons.run_proc('twitstream', {:dir_mode => :normal, :dir => PID_HOME, :multiple => false, :log_output => true}) do

  # daemonising changes the pwd to /, so we need to switch back to RAILS_HOME
  Dir.chdir(RAILS_HOME)
  require File.expand_path( RAILS_HOME + '/config/environment.rb' )

  EventMachine::run {
    $stdout.print "Starting TwitStream (#{RAILS_ENV})\n"
    stream = Twitter::JSONStream.connect(@twitconfig.symbolize_keys)
    stream.each_item do |tweet_json|
      tweet_hash = ActiveSupport::JSON.decode(tweet_json)
      $stdout.print "Incoming Tweet: #{tweet_hash['id']}\n"
      
      # uncomment this line if you want your Mac to 'say' the tweet
      #`say 'tweet from twit; #{tweet_hash['user']['screen_name']}; rambling on about; #{tweet_hash['text'].gsub(/\"|\'/, '')}'`

      tweet_user_attributes = { :name              => tweet_hash['user']['name'],
                                :screen_name       => tweet_hash['user']['screen_name'],
                                :location          => tweet_hash['user']['location'],
                                :url               => tweet_hash['user']['url'],
                                :profile_image_url => tweet_hash['user']['profile_image_url'],
                                :twitter_id        => tweet_hash['user']['id'] }

      tweet_attributes = { :content =>  tweet_hash['text'],
                           :source => tweet_hash['source'],
                           :twitter_id => tweet_hash['id'],
                           :twitter_date_time => Time.parse(tweet_hash['created_at']) }
      #find or create tweet user
      @tweet_user = TweetUser.find_by_twitter_id(tweet_user_attributes[:twitter_id])
      if @tweet_user
       @tweet_user.update_attributes!(tweet_user_attributes)
      else
       @tweet_user = TweetUser.create!(tweet_user_attributes)
      end
      @tweet_user.tweets.create!(tweet_attributes) unless Tweet.find_by_twitter_id(tweet_attributes[:twitter_id])

      $stdout.print "Tweets: #{Tweet.count}\n"
      $stdout.print "Tweet Users: #{TweetUser.count}\n"
      $stdout.flush
    end

    stream.on_error do |message|
      $stdout.print "TwitStream - error: #{message}\n"
      $stdout.flush
    end

    stream.on_reconnect do |timeout, retries|
      $stdout.print "TwitStream - reconnecting in: #{timeout} seconds\n"
      $stdout.flush
    end

    stream.on_max_reconnects do |timeout, retries|
      $stdout.print "TwitStream - failed after #{retries} failed reconnects\n"
      $stdout.flush
    end

    trap('TERM') {
      stream.stop
      EventMachine.stop if EventMachine.reactor_running?
    }
  }

  puts "\nFinished TwitStream (#{RAILS_ENV})"
  exit 0
end