<commit_before>/*
 * Copyright © 2012 The University of Texas at Dallas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package edu.utdallas.cs.stormrider.examples.twitter.add;

import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import com.hp.hpl.jena.sparql.vocabulary.FOAF;
import com.hp.hpl.jena.vocabulary.OWL;
import com.hp.hpl.jena.vocabulary.VCARD;

import edu.utdallas.cs.stormrider.util.TwitterConstants;
import edu.utdallas.cs.stormrider.views.Views;
import edu.utdallas.cs.stormrider.views.ViewsFactory;
import edu.utdallas.cs.stormrider.vocab.TWITTER;

import twitter4j.IDs;
import twitter4j.Paging;
import twitter4j.Status;
import twitter4j.Twitter;
import twitter4j.TwitterException;
import twitter4j.TwitterFactory;
import twitter4j.User;
import twitter4j.auth.AccessToken;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

/**
 * A Spout for emitting tuples from the Twitter data stream
 */
public class TwitterLoaderSpout implements IRichSpout 
{
	/** A default serial version uid **/
	private static final long serialVersionUID = 1L ;

	/** A Logger for this class **/
    private static Logger LOG = Logger.getLogger( TwitterLoaderSpout.class ) ;
    
    /** A variable that denotes if this spout is distributed **/
    private boolean isDistributed = false ;
    
    /** An output collector used to emit tuples from the Twitter stream **/
    private SpoutOutputCollector collector ;

    /** The OAuth verified Twitter handle **/
    private Twitter twitter = null ;
    
    private Views views = null ;
    
    /** Constructor **/
    public TwitterLoaderSpout( String viewsConfigFile ) { this( true, viewsConfigFile ) ; }

    /** Constructor **/
    public TwitterLoaderSpout( boolean isDistributed, String viewsConfigFile ) 
    { 
    	this.isDistributed = isDistributed ;
		twitter = new TwitterFactory().getInstance() ;
		twitter.setOAuthConsumer( TwitterConstants.CONSUMER_KEY, TwitterConstants.CONSUMER_SECRET ) ;
		AccessToken atoken = new AccessToken( TwitterConstants.ACCESS_TOKEN, TwitterConstants.ACCESS_TOKEN_SECRET ) ;
		twitter.setOAuthAccessToken( atoken ) ;
		this.views = ViewsFactory.getViews( viewsConfigFile ) ;
    }
    
    @Override
    public boolean isDistributed() { return isDistributed ; }
    
	@SuppressWarnings("rawtypes")
	@Override
    public void open( Map conf, TopologyContext context, SpoutOutputCollector collector ) { this.collector = collector ; }

    @Override
    public void close() { }
    
    @Override
    public void nextTuple() 
    {
    	Random random = new Random() ;
    	long numOfUsers = 0L ;
    	try
    	{
    		while( true )
    		{
    			//Increment numOfUsers
    			numOfUsers++ ;
    			
    			//Emit user-specific information
    			int randomUserId = random.nextInt( 400000000 ) ;
    			emitUserAndTweetInformation( randomUserId ) ;
    			
    			//Emit user-id for updating node-centric view
    			String tupleType = "node-nv" ;
    			StringBuilder subject = new StringBuilder( "u-" ) ; subject.append( randomUserId ) ;
    			collector.emit( new Values( tupleType, subject.toString(), "", "" ) ) ;
    			
    			//Emit info for updating landmarks-centric view
    			if( ( numOfUsers % 100000 ) == 0 )
    			{
        			tupleType = "node-lv" ;
    				Iterator<Result> iterTable = views.getAllRows() ;
    				while( iterTable.hasNext() )
    				{
    					Result row = iterTable.next() ;
    					Iterator<byte[]> iterAdjList = views.getAdjList( Bytes.toString( row.getRow() ) ) ;
    					while( iterAdjList.hasNext() )
    						collector.emit( new Values( tupleType, Bytes.toString( iterAdjList.next() ), "", "" ) ) ;
    				}
    			}
    		}
    	}
    	catch( Exception e ) { LOG.info( "Error in retrieving user information", e ) ; }
    }
    
    @Override
    public void ack( Object msgId ) { }

    @Override
    public void fail( Object msgId ) { }
    
    @Override
    public void declareOutputFields( OutputFieldsDeclarer declarer ) 
    {
    	declarer.declare( new Fields( "tupleType" ) ) ;
        declarer.declare( new Fields( "subject" ) ) ;
        declarer.declare( new Fields( "predicate" ) ) ;
        declarer.declare( new Fields( "object" ) ) ;
    }
    <commit_msg>Removed periodic part of loop for landmarks updation, since we need a rolling count<commit_after>/*
 * Copyright © 2012 The University of Texas at Dallas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package edu.utdallas.cs.stormrider.examples.twitter.add;

import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.log4j.Logger;

import com.hp.hpl.jena.sparql.vocabulary.FOAF;
import com.hp.hpl.jena.vocabulary.OWL;
import com.hp.hpl.jena.vocabulary.VCARD;

import edu.utdallas.cs.stormrider.store.Store;
import edu.utdallas.cs.stormrider.store.StoreFactory;
import edu.utdallas.cs.stormrider.store.iterator.impl.NodeAndNeighbor;
import edu.utdallas.cs.stormrider.util.TwitterConstants;
import edu.utdallas.cs.stormrider.views.Views;
import edu.utdallas.cs.stormrider.views.ViewsFactory;
import edu.utdallas.cs.stormrider.vocab.TWITTER;

import twitter4j.IDs;
import twitter4j.Paging;
import twitter4j.Status;
import twitter4j.Twitter;
import twitter4j.TwitterException;
import twitter4j.TwitterFactory;
import twitter4j.User;
import twitter4j.auth.AccessToken;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

/**
 * A Spout for emitting tuples from the Twitter data stream
 */
public class TwitterLoaderSpout implements IRichSpout 
{
	/** A default serial version uid **/
	private static final long serialVersionUID = 1L ;

	/** A Logger for this class **/
    private static Logger LOG = Logger.getLogger( TwitterLoaderSpout.class ) ;
    
    /** A variable that denotes if this spout is distributed **/
    private boolean isDistributed = false ;
    
    /** An output collector used to emit tuples from the Twitter stream **/
    private SpoutOutputCollector collector ;

    /** The OAuth verified Twitter handle **/
    private Twitter twitter = null ;
    
    private Store store = null ;
    
    private Views views = null ;
    
    /** Constructor **/
    public TwitterLoaderSpout( boolean isReified, String storeConfigFile, String viewsConfigFile ) { this( false, isReified, storeConfigFile, viewsConfigFile ) ; }

    /** Constructor **/
    public TwitterLoaderSpout( boolean isDistributed, boolean isReified, String storeConfigFile, String viewsConfigFile ) 
    { 
    	this.isDistributed = isDistributed ;
		twitter = new TwitterFactory().getInstance() ;
		twitter.setOAuthConsumer( TwitterConstants.CONSUMER_KEY, TwitterConstants.CONSUMER_SECRET ) ;
		AccessToken atoken = new AccessToken( TwitterConstants.ACCESS_TOKEN, TwitterConstants.ACCESS_TOKEN_SECRET ) ;
		twitter.setOAuthAccessToken( atoken ) ;
		this.store = StoreFactory.getJenaHBaseStore( storeConfigFile, isReified ) ;
		this.views = ViewsFactory.getViews( viewsConfigFile ) ;
    }
    
    @Override
    public boolean isDistributed() { return isDistributed ; }
    
	@SuppressWarnings("rawtypes")
	@Override
    public void open( Map conf, TopologyContext context, SpoutOutputCollector collector ) { this.collector = collector ; }

    @Override
    public void close() { }
    
    @Override
    public void nextTuple() 
    {
    	Random random = new Random() ;
    	long numOfUsers = 0L ;
    	try
    	{
    		while( true )
    		{
    			//Increment numOfUsers
    			numOfUsers++ ;
    			
    			//Emit user-specific information
    			int randomUserId = random.nextInt( 400000000 ) ;
    			emitUserAndTweetInformation( randomUserId ) ;
    			
    			//Emit user-id for updating node-centric view
    			String tupleType = "nv" ;
    			StringBuilder subject = new StringBuilder( "u-" ) ; subject.append( randomUserId ) ;
    			collector.emit( new Values( tupleType, subject.toString(), "", "" ) ) ;
    			
    			//Emit info for updating landmarks-centric view
       			tupleType = "nlv" ;
       			Iterator<NodeAndNeighbor> iterNodesAndNeighbors = store.getAllNodesWithNeighbors( views.getLinkNameAsURI() ) ;
       			while( iterNodesAndNeighbors.hasNext() )
       			{
       				NodeAndNeighbor nodeAndNeighbor = iterNodesAndNeighbors.next() ;
       				collector.emit( new Values( tupleType, nodeAndNeighbor.getNode(), "", "" ) ) ;
       			}
    		}
    	}
    	catch( Exception e ) { LOG.info( "Error in retrieving user information", e ) ; }
    }
    
    @Override
    public void ack( Object msgId ) { }

    @Override
    public void fail( Object msgId ) { }
    
    @Override
    public void declareOutputFields( OutputFieldsDeclarer declarer ) 
    {
    	declarer.declare( new Fields( "tupleType" ) ) ;
        declarer.declare( new Fields( "node1" ) ) ;
        declarer.declare( new Fields( "node2" ) ) ;
        declarer.declare( new Fields( "node3" ) ) ;
    }
    
{'fraction_non_alphanumeric': 0.09553231939163498, 'fraction_numerical': 0.006273764258555133, 'mean_word_length': 3.26815415821501, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6276923', 'n_tokens_mistral': 3379, 'n_tokens_neox': 3177, 'n_words': 834}