// Path: commons/src/at/illecker/storm/commons/util/TimeUtils.java
// public class TimeUtils {
// 
//   public static void sleepMillis(long millis) {
//     try {
//       Time.sleep(millis);
//     } catch (InterruptedException e) {
//       throw new RuntimeException(e);
//     }
//   }
// 
//   // 1 ms = 1 000 000 ns
//   public static void sleepNanos(long nanos) {
//     long start = System.nanoTime();
//     long end = 0;
//     do {
//       end = System.nanoTime();
//     } while (start + nanos >= end);
//     // System.out.println("waited time: " + (end - start) + " ns");
//   }
// }


import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import twitter4j.FilterQuery;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.auth.AccessToken;
import twitter4j.conf.ConfigurationBuilder;
import at.illecker.storm.commons.util.TimeUtils;
import backtype.storm.Config;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
  private String[] m_keyWords;
  private String m_filterLanguage;

  public TwitterStreamSpout(String consumerKey, String consumerSecret,
      String accessToken, String accessTokenSecret, String[] keyWords,
      String filterLanguage) {
    this.m_consumerKey = consumerKey;
    this.m_consumerSecret = consumerSecret;
    this.m_accessToken = accessToken;
    this.m_accessTokenSecret = accessTokenSecret;
    this.m_keyWords = keyWords;
    this.m_filterLanguage = filterLanguage; // "en"
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    // key of output tuples
    declarer.declare(new Fields("id", "text", "score"));
  }

  @Override
  public void open(Map config, TopologyContext context,
      SpoutOutputCollector collector) {
    m_collector = collector;
    m_tweetsQueue = new LinkedBlockingQueue<Status>(1000);

    // Optional startup sleep to finish bolt preparation
    // before spout starts emitting
    if (config.get(CONF_STARTUP_SLEEP_MS) != null) {
      long startupSleepMillis = (Long) config.get(CONF_STARTUP_SLEEP_MS);
// Path: commons/src/at/illecker/storm/commons/util/TimeUtils.java
// public class TimeUtils {
// 
//   public static void sleepMillis(long millis) {
//     try {
//       Time.sleep(millis);
//     } catch (InterruptedException e) {
//       throw new RuntimeException(e);
//     }
//   }
// 
//   // 1 ms = 1 000 000 ns
//   public static void sleepNanos(long nanos) {
//     long start = System.nanoTime();
//     long end = 0;
//     do {
//       end = System.nanoTime();
//     } while (start + nanos >= end);
//     // System.out.println("waited time: " + (end - start) + " ns");
//   }
// }


// Path: commons/src/at/illecker/storm/commons/spout/TwitterStreamSpout.java
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import twitter4j.FilterQuery;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.auth.AccessToken;
import twitter4j.conf.ConfigurationBuilder;
import at.illecker.storm.commons.util.TimeUtils;
import backtype.storm.Config;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

  private String[] m_keyWords;
  private String m_filterLanguage;

  public TwitterStreamSpout(String consumerKey, String consumerSecret,
      String accessToken, String accessTokenSecret, String[] keyWords,
      String filterLanguage) {
    this.m_consumerKey = consumerKey;
    this.m_consumerSecret = consumerSecret;
    this.m_accessToken = accessToken;
    this.m_accessTokenSecret = accessTokenSecret;
    this.m_keyWords = keyWords;
    this.m_filterLanguage = filterLanguage; // "en"
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    // key of output tuples
    declarer.declare(new Fields("id", "text", "score"));
  }

  @Override
  public void open(Map config, TopologyContext context,
      SpoutOutputCollector collector) {
    m_collector = collector;
    m_tweetsQueue = new LinkedBlockingQueue<Status>(1000);

    // Optional startup sleep to finish bolt preparation
    // before spout starts emitting
    if (config.get(CONF_STARTUP_SLEEP_MS) != null) {
      long startupSleepMillis = (Long) config.get(CONF_STARTUP_SLEEP_MS);
      TimeUtils.sleepMillis(startupSleepMillis);