feeder.py

#!/usr/bin/python


import base64
import logging
import Queue
import sys
import threading
import time
import urllib2


def producer(request, q):
  while True:
    retry_delay = 0.2
    try:
      response = urllib2.urlopen(request)
      retry_delay = 0.2
      for line in response:
        q.put(line)
    except (urllib2.URLError, urllib2.HTTPError), e:
      logging.exception(
          "Received exception while fetching, retrying in %d seconds",
          retry_interval)
      time.sleep(retry_delay)
      retry_delay *= 2.0


def upload_batch(batch, post_to):
  request = urllib2.Request(
      post_to,
      ''.join(batch),
      {'Content-Type': 'application/json'})
  retry_delay = 0.2
  while True:
    try:
      urllib2.urlopen(request)
      break
    except (urllib2.URLError, urllib2.HTTPError), e:
      logging.exception(
          "Received exception while uploading data, retrying in %d seconds",
          retry_interval)
      time.sleep(retry_delay)
      retry_delay *= 2.0


def consumer(q, post_to, batch_size=100, max_wait=60.0):
  batch = []
  now = time.time()
  wait_until = now + max_wait
  while True:
    if now >= wait_until or len(batch) >= batch_size:
      upload_batch(batch, post_to)
      batch = []
      now = time.time()
      wait_until = now + max_wait
    try:
      line = q.get(True, wait_until - now)
      batch.append(line)
    except Queue.Empty:
      pass
    now = time.time()


def main(url, auth, post_to):
  q = Queue.Queue()
  consumer_thread = threading.Thread(target=consumer, args=(q, post_to))
  consumer_thread.start()

  auth = base64.b64encode(auth)
  twitter_req = urllib2.Request(
      url,
      None,
      {'Authorization': 'Basic %s' % (auth,)})
  producer(twitter_req, q)


if __name__ == '__main__':
  logging.basicConfig(level=logging.DEBUG)
  main(*sys.argv[1:])