<commit_before>/**
 * 
 */
package edu.brandeis.flow.core.operator.in.twitter;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.json.JSONException;

import edu.brandeis.flow.core.json.JSONObject;

import com.google.common.collect.Lists;
import com.twitter.hbc.ClientBuilder;
import com.twitter.hbc.core.Client;
import com.twitter.hbc.core.Constants;
import com.twitter.hbc.core.Hosts;
import com.twitter.hbc.core.HttpHosts;
import com.twitter.hbc.core.endpoint.StatusesFilterEndpoint;
import com.twitter.hbc.core.endpoint.StatusesFirehoseEndpoint;
import com.twitter.hbc.core.endpoint.StatusesSampleEndpoint;
import com.twitter.hbc.core.event.Event;
import com.twitter.hbc.core.processor.StringDelimitedProcessor;
import com.twitter.hbc.httpclient.auth.Authentication;
import com.twitter.hbc.httpclient.auth.OAuth1;

import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterObjectFactory;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.conf.ConfigurationBuilder;
import edu.brandeis.flow.core.operator.JSONOperator;
import edu.brandeis.flow.server.source.Streamer;
import edu.brandeis.flow.server.source.twitter.TwitterStreamSource;
import edu.brandeis.flow.server.stream.JSONThread;

/**
 * @author Yahui
 *
 */
public class TwitterIN extends JSONOperator{

	Client hosebirdClient;
	BlockingQueue<String> msgQueue;
	StatusesFilterEndpoint hosebirdEndpoint;
	ClientBuilder builder;
	
	public TwitterIN() {
		super();
	}
	
	public void setup(){
		/** Set up your blocking queues: Be sure to size these properly based on expected TPS of your stream */
		msgQueue = new LinkedBlockingQueue<String>(100000);
		BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<Event>(1000);

		/** Declare the host you want to connect to, the endpoint, and authentication (basic auth or oauth) */
		Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
		StatusesSampleEndpoint endpoint = new StatusesSampleEndpoint();
		// Optional: set up some followings and track terms
		//List<Long> followings = Lists.newArrayList(1234L, 566788L);
		//hosebirdEndpoint.followings(followings);
		//hosebirdEndpoint.trackTerms(Lists.newArrayList("wow"));


		// These secrets should be read from a config file
		Authentication hosebirdAuth = new OAuth1("<KEY>", "<KEY>", "<KEY>", "<KEY>");
		builder = new ClientBuilder()
		  .name("Hosebird-Client-01")                              // optional: mainly for the logs
		  .hosts(hosebirdHosts)
		  .authentication(hosebirdAuth)
		  .endpoint(endpoint)
		  .processor(new StringDelimitedProcessor(msgQueue))
		  .eventMessageQueue(eventQueue);                          // optional: use this if you want to process client events
		hosebirdClient = builder.build();
		// Attempts to establish a connection.
		hosebirdClient.connect();
		
	}
	
	public void clear(){
		hosebirdClient.stop();
		setup();
	}
	
	public void resetParam(List<String> params){
		//hosebirdClient.stop();
		/** Set up your blocking queues: Be sure to size these properly based on expected TPS of your stream */
		msgQueue = new LinkedBlockingQueue<String>(100000);
		BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<Event>(1000);

		/** Declare the host you want to connect to, the endpoint, and authentication (basic auth or oauth) */
		Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
		hosebirdEndpoint = new StatusesFilterEndpoint();
		// Optional: set up some followings and track terms
		//List<Long> followings = Lists.newArrayList(1234L, 566788L);
		List<String> terms = params;
		//hosebirdEndpoint.followings(followings);
		hosebirdEndpoint.trackTerms(terms);
		

		// These secrets should be read from a config file
		
		Authentication hosebirdAuth = new OAuth1("<KEY>", "<KEY>", "<KEY>", "<KEY>");
		builder = new ClientBuilder()
		  .name("Hosebird-Client-01")                              // optional: mainly for the logs
		  .hosts(hosebirdHosts)
		  .authentication(hosebirdAuth)
		  .endpoint(hosebirdEndpoint)
		  .processor(new StringDelimitedProcessor(msgQueue))
		  .eventMessageQueue(eventQueue);                          // optional: use this if you want to process client events
		hosebirdClient = builder.build();
		// Attempts to establish a connection.
		hosebirdClient.connect();
		
	}
	
	

	
	

	@Override
	public void run() {
		setup();

		String msg;
		while (true) {
			  
			try {
				msg = msgQueue.take();
				
				System.out.println(msg);
				send(new JSONObject(msg));
				
				
			} catch (InterruptedException e) {
				hosebirdClient.stop();
				break;
			} catch (JSONException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			  
		}
	}
	
	
	


}
<commit_msg>last min fix no println<commit_after>/**
 * 
 */
package edu.brandeis.flow.core.operator.in.twitter;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.json.JSONException;

import edu.brandeis.flow.core.json.JSONObject;

import com.google.common.collect.Lists;
import com.twitter.hbc.ClientBuilder;
import com.twitter.hbc.core.Client;
import com.twitter.hbc.core.Constants;
import com.twitter.hbc.core.Hosts;
import com.twitter.hbc.core.HttpHosts;
import com.twitter.hbc.core.endpoint.StatusesFilterEndpoint;
import com.twitter.hbc.core.endpoint.StatusesFirehoseEndpoint;
import com.twitter.hbc.core.endpoint.StatusesSampleEndpoint;
import com.twitter.hbc.core.event.Event;
import com.twitter.hbc.core.processor.StringDelimitedProcessor;
import com.twitter.hbc.httpclient.auth.Authentication;
import com.twitter.hbc.httpclient.auth.OAuth1;

import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterObjectFactory;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.conf.ConfigurationBuilder;
import edu.brandeis.flow.core.operator.JSONOperator;
import edu.brandeis.flow.server.source.Streamer;
import edu.brandeis.flow.server.source.twitter.TwitterStreamSource;
import edu.brandeis.flow.server.stream.JSONThread;

/**
 * @author Yahui
 *
 */
public class TwitterIN extends JSONOperator{

	Client hosebirdClient;
	BlockingQueue<String> msgQueue;
	StatusesFilterEndpoint hosebirdEndpoint;
	ClientBuilder builder;
	
	public TwitterIN() {
		super();
	}
	
	public void setup(){
		/** Set up your blocking queues: Be sure to size these properly based on expected TPS of your stream */
		msgQueue = new LinkedBlockingQueue<String>(100000);
		BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<Event>(1000);

		/** Declare the host you want to connect to, the endpoint, and authentication (basic auth or oauth) */
		Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
		StatusesSampleEndpoint endpoint = new StatusesSampleEndpoint();
		// Optional: set up some followings and track terms
		//List<Long> followings = Lists.newArrayList(1234L, 566788L);
		//hosebirdEndpoint.followings(followings);
		//hosebirdEndpoint.trackTerms(Lists.newArrayList("wow"));


		// These secrets should be read from a config file
		Authentication hosebirdAuth = new OAuth1("<KEY>", "<KEY>", "<KEY>", "<KEY>");
		builder = new ClientBuilder()
		  .name("Hosebird-Client-01")                              // optional: mainly for the logs
		  .hosts(hosebirdHosts)
		  .authentication(hosebirdAuth)
		  .endpoint(endpoint)
		  .processor(new StringDelimitedProcessor(msgQueue))
		  .eventMessageQueue(eventQueue);                          // optional: use this if you want to process client events
		hosebirdClient = builder.build();
		// Attempts to establish a connection.
		hosebirdClient.connect();
		
	}
	
	public void clear(){
		hosebirdClient.stop();
		setup();
	}
	
	public void resetParam(List<String> params){
		//hosebirdClient.stop();
		/** Set up your blocking queues: Be sure to size these properly based on expected TPS of your stream */
		msgQueue = new LinkedBlockingQueue<String>(100000);
		BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<Event>(1000);

		/** Declare the host you want to connect to, the endpoint, and authentication (basic auth or oauth) */
		Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
		hosebirdEndpoint = new StatusesFilterEndpoint();
		// Optional: set up some followings and track terms
		//List<Long> followings = Lists.newArrayList(1234L, 566788L);
		List<String> terms = params;
		//hosebirdEndpoint.followings(followings);
		hosebirdEndpoint.trackTerms(terms);
		

		// These secrets should be read from a config file
		
		Authentication hosebirdAuth = new OAuth1("<KEY>", "<KEY>", "<KEY>", "<KEY>");
		builder = new ClientBuilder()
		  .name("Hosebird-Client-01")                              // optional: mainly for the logs
		  .hosts(hosebirdHosts)
		  .authentication(hosebirdAuth)
		  .endpoint(hosebirdEndpoint)
		  .processor(new StringDelimitedProcessor(msgQueue))
		  .eventMessageQueue(eventQueue);                          // optional: use this if you want to process client events
		hosebirdClient = builder.build();
		// Attempts to establish a connection.
		hosebirdClient.connect();
		
	}
	
	

	
	

	@Override
	public void run() {
		setup();

		String msg;
		while (true) {
			  
			try {
				msg = msgQueue.take();
				
				//System.out.println(msg);
				send(new JSONObject(msg));
				
				
			} catch (InterruptedException e) {
				hosebirdClient.stop();
				break;
			} catch (JSONException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			  
		}
	}
	
	
	


}

{'fraction_non_alphanumeric': 0.10786106032906764, 'fraction_numerical': 0.011172049563274427, 'mean_word_length': 3.975745325922183, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5531708', 'n_tokens_mistral': 3325, 'n_tokens_neox': 2916, 'n_words': 708}