src/main/java/br/org/soujava/twitter/controller/TwitterConnect.java
private void processTwitts(InputStream source, int maxMgsPerTime, int TimeoutFollowing, String stringToFind) {

        logger.debug("Start Following, {} twitters, {} seconds", maxMgsPerTime, TimeoutFollowing);

        Date dateStarter = new java.util.Date();

        LinkedBlockingQueue<String> messageQueue = new LinkedBlockingQueue<>(maxMgsPerTime);

        CallableThread tweetsCallableThread = new CallableThread(source, messageQueue, maxMgsPerTime, TimeoutFollowing);

        Future<Boolean> messageReaderResult = executorServiceMessagePool.submit(tweetsCallableThread);

        Unmarshaller tweetsUnmarshaller = new Unmarshaller(messageQueue);
        Future<List<Twitter>> messageDeserializationResult = executorServiceMessageDeserializePool.submit(tweetsUnmarshaller);

        Integer TOTAL_ALL_ROUNDS = 0;

        Integer ROUND_COUNT = 0;

        Double TOTAL_MESSEGES_SECOND = new Double(0);
        DB db = DBMaker.fileDB("stats.db").make();
        try {

            ConcurrentMap statistics = db.hashMap("STATISTICS").createOrOpen();

            boolean receivedAllMessages = messageReaderResult.get();

            if (receivedAllMessages) {

                Date dateFinished = new java.util.Date();

                long totalSecond = getDateDiff(dateStarter, dateFinished, TimeUnit.SECONDS);

                tweetsUnmarshaller.concludeProcessing();
                List<Twitter> deSerializedTweets = messageDeserializationResult.get();
                twitters.addAll(deSerializedTweets);

                Integer TOTAL_THIS_ROUND = deSerializedTweets.size();

                // ** PRINT STATISTICS ** //

                Object TOTAL_ALL_ROUNDS_OBJ = statistics.get("TOTAL_ALL_ROUNDS");

                Object ROUND_COUNT_OBJ = statistics.get("ROUND_COUNT");

                if (ROUND_COUNT_OBJ instanceof Integer) {
                    ROUND_COUNT = (Integer) ROUND_COUNT_OBJ;
                }

                if (TOTAL_ALL_ROUNDS_OBJ instanceof Integer) {
                    TOTAL_ALL_ROUNDS = (Integer) TOTAL_ALL_ROUNDS_OBJ;
                }

                TOTAL_ALL_ROUNDS = TOTAL_ALL_ROUNDS + TOTAL_THIS_ROUND;
                statistics.put("TOTAL_ALL_ROUNDS", TOTAL_ALL_ROUNDS);

                ROUND_COUNT = ROUND_COUNT + 1;
                statistics.put("ROUND_COUNT", ROUND_COUNT);


                TOTAL_MESSEGES_SECOND = TOTAL_THIS_ROUND.doubleValue() / totalSecond; //new Long(totalSecond) /new Long(TOTAL_THIS_ROUND.intValue()) ;

                String pattern = "###.##";
                DecimalFormat decimalFormat = new DecimalFormat(pattern);

                ObjectNode objectNode1 = mapper.createObjectNode();
                objectNode1.put("TOTAL_THIS_ROUND", TOTAL_THIS_ROUND);
                objectNode1.put("TOTAL_ALL_ROUNDS", TOTAL_ALL_ROUNDS);
                objectNode1.put("TOTAL_MESSAGES_PER_SECOND", decimalFormat.format(TOTAL_MESSEGES_SECOND));
                objectNode1.put("ROUND_COUNT", ROUND_COUNT);
                objectNode1.put("FOLLOW.KEYWORDS", stringToFind);

                logger.info(mapper.writer().writeValueAsString(objectNode1));

            }

        } catch (Exception e) {
            logger.error("Error: following(). Aborting tracking!", e);
        } finally {

            logger.debug("Shutting down....");
            executorServiceMessagePool.shutdown();

            logger.debug("Shutting down....");
            executorServiceMessageDeserializePool.shutdown();
            db.close();

        }
    }
/**
     * Start Following for X tweets for X seconds
     *
     * @param source
     * @param maxMgsPerTime
     * @param TimeoutFollowing
     */
Start Following for X tweets for X seconds
@param source
@param maxMgsPerTime
@param TimeoutFollowing
Start Following for X tweets for X seconds @param source @param maxMgsPerTime @param TimeoutFollowing
private void processTwitts(InputStream source, int maxMgsPerTime, int TimeoutFollowing, String stringToFind) {
        logger.debug("Start Following, {} twitters, {} seconds", maxMgsPerTime, TimeoutFollowing);
        Date dateStarter = new java.util.Date();
        LinkedBlockingQueue<String> messageQueue = new LinkedBlockingQueue<>(maxMgsPerTime);
        CallableThread tweetsCallableThread = new CallableThread(source, messageQueue, maxMgsPerTime, TimeoutFollowing);
        Future<Boolean> messageReaderResult = executorServiceMessagePool.submit(tweetsCallableThread);
        Unmarshaller tweetsUnmarshaller = new Unmarshaller(messageQueue);
        Future<List<Twitter>> messageDeserializationResult = executorServiceMessageDeserializePool.submit(tweetsUnmarshaller);
        Integer TOTAL_ALL_ROUNDS = 0;
        Integer ROUND_COUNT = 0;
        Double TOTAL_MESSEGES_SECOND = new Double(0);
        DB db = DBMaker.fileDB("stats.db").make();
        try {
            ConcurrentMap statistics = db.hashMap("STATISTICS").createOrOpen();
            boolean receivedAllMessages = messageReaderResult.get();
            if (receivedAllMessages) {
                Date dateFinished = new java.util.Date();
                long totalSecond = getDateDiff(dateStarter, dateFinished, TimeUnit.SECONDS);
                tweetsUnmarshaller.concludeProcessing();
                List<Twitter> deSerializedTweets = messageDeserializationResult.get();
                twitters.addAll(deSerializedTweets);
                Integer TOTAL_THIS_ROUND = deSerializedTweets.size();
                Object TOTAL_ALL_ROUNDS_OBJ = statistics.get("TOTAL_ALL_ROUNDS");
                Object ROUND_COUNT_OBJ = statistics.get("ROUND_COUNT");
                if (ROUND_COUNT_OBJ instanceof Integer) {
                    ROUND_COUNT = (Integer) ROUND_COUNT_OBJ;
                }
                if (TOTAL_ALL_ROUNDS_OBJ instanceof Integer) {
                    TOTAL_ALL_ROUNDS = (Integer) TOTAL_ALL_ROUNDS_OBJ;
                }
                TOTAL_ALL_ROUNDS = TOTAL_ALL_ROUNDS + TOTAL_THIS_ROUND;
                statistics.put("TOTAL_ALL_ROUNDS", TOTAL_ALL_ROUNDS);
                ROUND_COUNT = ROUND_COUNT + 1;
                statistics.put("ROUND_COUNT", ROUND_COUNT);
                TOTAL_MESSEGES_SECOND = TOTAL_THIS_ROUND.doubleValue() / totalSecond; 
                String pattern = "###.##";
                DecimalFormat decimalFormat = new DecimalFormat(pattern);
                ObjectNode objectNode1 = mapper.createObjectNode();
                objectNode1.put("TOTAL_THIS_ROUND", TOTAL_THIS_ROUND);
                objectNode1.put("TOTAL_ALL_ROUNDS", TOTAL_ALL_ROUNDS);
                objectNode1.put("TOTAL_MESSAGES_PER_SECOND", decimalFormat.format(TOTAL_MESSEGES_SECOND));
                objectNode1.put("ROUND_COUNT", ROUND_COUNT);
                objectNode1.put("FOLLOW.KEYWORDS", stringToFind);
                logger.info(mapper.writer().writeValueAsString(objectNode1));
            }
        } catch (Exception e) {
            logger.error("Error: following(). Aborting tracking!", e);
        } finally {
            logger.debug("Shutting down....");
            executorServiceMessagePool.shutdown();
            logger.debug("Shutting down....");
            executorServiceMessageDeserializePool.shutdown();
            db.close();
        }
    }
private void processTwitts ( InputStream source , int maxMgsPerTime , int TimeoutFollowing , String stringToFind ) { logger . debug ( "Start Following, {} twitters, {} seconds" , maxMgsPerTime , TimeoutFollowing ) ; Date dateStarter = new java . util . Date ( ) ; LinkedBlockingQueue < String > messageQueue = new LinkedBlockingQueue < > ( maxMgsPerTime ) ; CallableThread tweetsCallableThread = new CallableThread ( source , messageQueue , maxMgsPerTime , TimeoutFollowing ) ; Future < Boolean > messageReaderResult = executorServiceMessagePool . submit ( tweetsCallableThread ) ; Unmarshaller tweetsUnmarshaller = new Unmarshaller ( messageQueue ) ; Future < List < Twitter > > messageDeserializationResult = executorServiceMessageDeserializePool . submit ( tweetsUnmarshaller ) ; Integer TOTAL_ALL_ROUNDS = 0 ; Integer ROUND_COUNT = 0 ; Double TOTAL_MESSEGES_SECOND = new Double ( 0 ) ; DB db = DBMaker . fileDB ( "stats.db" ) . make ( ) ; try { ConcurrentMap statistics = db . hashMap ( "STATISTICS" ) . createOrOpen ( ) ; boolean receivedAllMessages = messageReaderResult . get ( ) ; if ( receivedAllMessages ) { Date dateFinished = new java . util . Date ( ) ; long totalSecond = getDateDiff ( dateStarter , dateFinished , TimeUnit . SECONDS ) ; tweetsUnmarshaller . concludeProcessing ( ) ; List < Twitter > deSerializedTweets = messageDeserializationResult . get ( ) ; twitters . addAll ( deSerializedTweets ) ; Integer TOTAL_THIS_ROUND = deSerializedTweets . size ( ) ; Object TOTAL_ALL_ROUNDS_OBJ = statistics . get ( "TOTAL_ALL_ROUNDS" ) ; Object ROUND_COUNT_OBJ = statistics . get ( "ROUND_COUNT" ) ; if ( ROUND_COUNT_OBJ instanceof Integer ) { ROUND_COUNT = ( Integer ) ROUND_COUNT_OBJ ; } if ( TOTAL_ALL_ROUNDS_OBJ instanceof Integer ) { TOTAL_ALL_ROUNDS = ( Integer ) TOTAL_ALL_ROUNDS_OBJ ; } TOTAL_ALL_ROUNDS = TOTAL_ALL_ROUNDS + TOTAL_THIS_ROUND ; statistics . put ( "TOTAL_ALL_ROUNDS" , TOTAL_ALL_ROUNDS ) ; ROUND_COUNT = ROUND_COUNT + 1 ; statistics . put ( "ROUND_COUNT" , ROUND_COUNT ) ; TOTAL_MESSEGES_SECOND = TOTAL_THIS_ROUND . doubleValue ( ) / totalSecond ; String pattern = "###.##" ; DecimalFormat decimalFormat = new DecimalFormat ( pattern ) ; ObjectNode objectNode1 = mapper . createObjectNode ( ) ; objectNode1 . put ( "TOTAL_THIS_ROUND" , TOTAL_THIS_ROUND ) ; objectNode1 . put ( "TOTAL_ALL_ROUNDS" , TOTAL_ALL_ROUNDS ) ; objectNode1 . put ( "TOTAL_MESSAGES_PER_SECOND" , decimalFormat . format ( TOTAL_MESSEGES_SECOND ) ) ; objectNode1 . put ( "ROUND_COUNT" , ROUND_COUNT ) ; objectNode1 . put ( "FOLLOW.KEYWORDS" , stringToFind ) ; logger . info ( mapper . writer ( ) . writeValueAsString ( objectNode1 ) ) ; } } catch ( Exception e ) { logger . error ( "Error: following(). Aborting tracking!" , e ) ; } finally { logger . debug ( "Shutting down...." ) ; executorServiceMessagePool . shutdown ( ) ; logger . debug ( "Shutting down...." ) ; executorServiceMessageDeserializePool . shutdown ( ) ; db . close ( ) ; } }
Start Following for X tweets for X seconds
 @param source
 @param maxMgsPerTime
 @param TimeoutFollowing
Start Following for X tweets for X seconds @param source @param maxMgsPerTime @param TimeoutFollowing
// ** PRINT STATISTICS ** // //new Long(totalSecond) /new Long(TOTAL_THIS_ROUND.intValue()) ;