package com.gopivotal.flume.source;

import org.apache.flume.Context;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.auth.OAuthAuthorization;
import twitter4j.conf.ConfigurationBuilder;

public class TwitterSampleStreamSource extends AbstractSource implements
		EventDrivenSource, Configurable {

	private static final Logger LOG = LoggerFactory
			.getLogger(TwitterSampleStreamSource.class);

	private String consumerKey;
	private String consumerSecret;
	private String accessToken;
	private String accessTokenSecret;
	private String authUrl;
	private String accessTokenUrl;
	private String requestTokenUrl;

	private TwitterStream stream = null;;

	@Override
	public void configure(Context context) {

		// Get all necessary parameters from the Flume configuration
		consumerKey = context.getString("consumer.key");
		consumerSecret = context.getString("consumer.secret");
		accessToken = context.getString("access.token");
		accessTokenSecret = context.getString("access.token.secret");
		authUrl = context.getString("access.token.url",
				"https://api.twitter.com/oauth/request_token");
		accessTokenUrl = context.getString("auth.url",
				"https://api.twitter.com/oauth/authorize");
		requestTokenUrl = context.getString("request.token.url",
				"https://api.twitter.com/oauth/access_token");

		// Log the properties
		LOG.info("Properties: {}", context.getParameters());
	}

	@Override
	public void start() {

		ChannelProcessor channel = getChannelProcessor();

		stream = getTwitterStream();

		// Add a new listener to the stream
		stream.addListener(new TwitterEventChannelListener(channel));

		LOG.info("Starting Twitter sample stream...");
		stream.sample();

		super.start();
	}

	@Override
	public void stop() {

		if (stream != null) {
			LOG.info("Shutting down Twitter stream");
			stream.shutdown();
		} else {
			LOG.info("Twitter stream is null");
		}

		super.stop();
	}

	/**
	 * Gets the configured Twitter Stream
	 * 
	 * @return the Twitter Stream
	 */
	private TwitterStream getTwitterStream() {

		ConfigurationBuilder twitterConf = new ConfigurationBuilder();

		twitterConf.setOAuthConsumerKey(consumerKey);
		twitterConf.setOAuthConsumerSecret(consumerSecret);

		twitterConf.setOAuthAccessToken(accessToken);
		twitterConf.setOAuthAccessTokenSecret(accessTokenSecret);

		twitterConf.setDebugEnabled(true);
		twitterConf.setJSONStoreEnabled(true);

		twitterConf.setOAuthAccessTokenURL(accessTokenUrl);
		twitterConf.setOAuthAuthorizationURL(authUrl);
		twitterConf.setOAuthRequestTokenURL(requestTokenUrl);

		OAuthAuthorization oauth = new OAuthAuthorization(twitterConf.build());

		return new TwitterStreamFactory().getInstance(oauth);
	}
}

@@ -304,16 +304,46 @@
 ctory;%0A%0A
+import twitter4j.FilterQuery;%0A
 import t
@@ -366,16 +366,16 @@
 Stream;%0A
-
 import t
@@ -914,16 +914,49 @@
 okenUrl;
+%0A%09private String%5B%5D filter = null;
 %0A%0A%09priva
@@ -1637,16 +1637,138 @@
 token%22);
+%0A%09%09String strFilter = context.getString(%22filter.by%22, null);%0A%0A%09%09filter = (strFilter != null) ? strFilter.split(%22,%22) : null;
 %0A%0A%09%09// L
@@ -2067,16 +2067,41 @@
 nel));%0A%0A
+%09%09if (filter == null) %7B%0A%09
 %09%09LOG.in
@@ -2143,16 +2143,17 @@
 ..%22);%0A%09%09
+%09
 stream.s
@@ -2160,16 +2160,190 @@
 ample();
+%0A%09%09%7D else %7B%0A%09%09%09LOG.info(%22Starting Twitter filtering on %7B%7D%22, (Object%5B%5D) filter);%0A%09%09%09FilterQuery query = new FilterQuery();%0A%09%09%09query.track(filter);%0A%09%09%09stream.filter(query);%0A%09%09%7D
 %0A%0A%09%09supe

Add filtering to source