otto.rb

require 'drb'
require 'pp'
require 'net/https'
require 'oauth'
require 'json'

class JSONStream
  def initialize(drop)
    @buf = ''
    @drop = drop
  end
  
  def push(str)
    @buf << str
    while (line = @buf[/.+?(\r\n)+/m]) != nil
      begin
        @buf.sub!(line,"")
        line.strip!
        event = JSON.parse(line)
      rescue
        break
      end
      @drop.write(event, 'DropDemo Event')
    end
  end
end

class Reader
  def initialize(drop)
    @drop = drop
    @place = nil
  end

  def oauth_token
    _, config, = @drop.older(nil, 'DropDemo OAuth')
    consumer = OAuth::Consumer.new(config[:consumer_key],
                                   config[:consumer_secret],
                                   :site => 'http://twitter.com')
    access_token = OAuth::AccessToken.new(consumer,
                                          config[:access_token],
                                          config[:access_token_secret])
    return consumer, access_token
  end

  def occupy
    @place = Thread.current
  end

  def occupied?
    @place == Thread.current
  end

  def fetch
    consumer, access_token = oauth_token

    uri = URI.parse('https://userstream.twitter.com/2/user.json')
    https = Net::HTTP.new(uri.host, uri.port)
    https.use_ssl = true
    https.ca_file = './verisign.cer'
    https.verify_mode = OpenSSL::SSL::VERIFY_PEER
    https.verify_depth = 5

    occupy
    https.start do |https|
      json = JSONStream.new(@drop)
      request = Net::HTTP::Get.new(uri.request_uri)
      request.oauth!(https, consumer, access_token)
      https.request(request) do |response|
        response.read_body do |chunk|
          return unless occupied?
          json.push(chunk)
        end
      end
    end
  end
end

#
#
MyDrop = DRbObject.new_with_uri('drbunix:' + File.expand_path('~/.drop/port'))

def main
  key = MyDrop.time_to_key(Time.now)
  while true
    MyDrop.read_tag(key, 'DropDemo Event', 10, 1).each do |k, v, tags|
      key = k
      pp v
    end
  end
end

def fetcher(reader)
  reader.fetch
end

reader = Reader.new(MyDrop)

Thread.new do
  main
end

while gets
  Thread.new do
    fetcher(reader)
  end
end