7
\$\begingroup\$

I’ve been experimenting with the Twitter Streaming API and would like some critical feedback on my latest project. Specifically code correctness, code smells, overall structure, and my usage of collections. The application identifies the current top trending hashtags for the supplied hashtag, or string.

AbstractClient.java

package com.gmail.lifeofreilly.lotus;

/**
 * An Abstract client for retrieving messages that contain hashtags. Can be extended for target social network.
 */
public abstract class AbstractClient implements Runnable {
    private String trackedTerm;
    private String screenName;
    private Long id;
    private MessageData messageData;

    public MessageData getMessageData() {
        return messageData;
    }

    public void setMessageData(final MessageData messageData) {
        this.messageData = messageData;
    }

    public String getTrackedTerm() {
        return trackedTerm;
    }

    public void setTrackedTerm(final String trackedTerm) {
        this.trackedTerm = trackedTerm;
    }

    public String getScreenName() {
        return screenName;
    }

    public void setScreenName(final String screenName) {
        this.screenName = screenName;
    }

    public Long getID() {
        return id;
    }

    public void setId(final Long id) {
        this.id = id;
    }

    @Override
    public String toString() {
        return "Client{" +
                "trackedTerm='" + trackedTerm + '\'' +
                ", screenName='" + screenName + '\'' +
                ", id=" + id +
                '}';
    }
}

MessageData.java

package com.gmail.lifeofreilly.lotus;

import org.apache.log4j.Logger;

import com.google.common.collect.Multiset;
import com.google.common.collect.Multisets;
import com.google.common.collect.TreeMultiset;

import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.Set;

/**
 * A message queue and the hashtags extracted.
 */
public class MessageData {
    private final static Logger log = Logger.getLogger(MessageData.class);
    private final Multiset<String> hashTags = TreeMultiset.create();
    private final Queue<String> messageQueue = new LinkedList<String>();
    private long messageCount;

    /**
     * Add a message to the queue to be processed.
     *
     * @param message the message.
     */
    public void addMessage(final String message) {
        messageQueue.add(message);
        messageCount++;
        log.debug("Current Queue size: " + messageQueue.size());
    }

    /**
     * Get the number of messages submitted for processing.
     *
     * @return the number of messages.
     */
    public long getMessageCount() {
        return messageCount;
    }

    /**
     * Returns true if the message queue is empty.
     *
     * @return is queue empty.
     */
    public boolean messageQueueIsEmpty() {
        return messageQueue.isEmpty();
    }

    /**
     * Returns and removes a message from the queue.
     *
     * @return the message.
     */
    public String removeMessageFromQueue() {
        return messageQueue.remove();
    }

    /**
     * Adds a hashtag to the collection.
     *
     * @param hashtag the hashtag.
     */
    public void addHashTag(final String hashtag) {
        hashTags.add(hashtag);
    }

    /**
     * Get the top hashtags.
     *
     * @return the top hashtags and occurrence of each.
     */
    public Map<String, Integer> getTopHashtags(int maxNumberOfHashTags) {
        Set<String> sortedSet = Multisets.copyHighestCountFirst(hashTags).elementSet();
        Iterator<String> iterator = sortedSet.iterator();
        Map<String, Integer> topTerms = new LinkedHashMap<String, Integer>();

        for (int i = 0; i < maxNumberOfHashTags; i++) {
            if (iterator.hasNext()) {
                String term = iterator.next();
                topTerms.put(term, hashTags.count(term));
            } else {
                break;
            }
        }

        return topTerms;
    }

}

MessageProcessor.java

package com.gmail.lifeofreilly.lotus;

import org.apache.log4j.Logger;

import java.util.StringTokenizer;

/**
 * Extracts hashtags from messages.
 */
public class MessageProcessor implements Runnable {
    private final static Logger log = Logger.getLogger(MessageProcessor.class);
    private final MessageData messageData;

    /**
     * Constructs a MessageProcessor.
     *
     * @param messageData the MessageData.
     */
    public MessageProcessor(final MessageData messageData) {
        this.messageData = messageData;
    }

    @Override
    public void run() {
        synchronized (messageData) {
            while (true) {
                if (!messageData.messageQueueIsEmpty()) {
                    log.debug("Extracting hashtags from message.");
                    extractHashtagsFromMessage(messageData.removeMessageFromQueue());
                } else {
                    log.debug("The queue is empty. Waiting...");
                    try {
                        messageData.wait();
                    } catch (InterruptedException ex) {
                        log.error("InterruptedException thrown: " + ex);
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }
    }

    private void extractHashtagsFromMessage(final String message) {
        String deliminator = " \t\n\r\f,.:;?![]'"; //adds punctuation marks to default set
        StringTokenizer tokenizer = new StringTokenizer(message, deliminator);
        while (tokenizer.hasMoreTokens()) {
            String token = tokenizer.nextToken();
            if (token.startsWith("#")) {
                messageData.addHashTag(token);
            }
        }
    }
}

TwitterClient.java

package com.gmail.lifeofreilly.lotus;

import org.apache.log4j.Logger;

import twitter4j.FilterQuery;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterException;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.TwitterFactory;

/**
 * Utilizes the Twitter Streaming API to collect messages.
 */
class TwitterClient extends AbstractClient {

    private final static Logger log = Logger.getLogger(TwitterClient.class);

    /**
     * Constructs a Twitter Client using the supplied MessageData object and tracked term.
     *
     * @param messageData the data structure for the Twitter data.
     * @param trackedTerm the term to track on Twitter.
     */
    public TwitterClient(final MessageData messageData, final String trackedTerm) {
        this.setMessageData(messageData);
        this.setTrackedTerm(trackedTerm);

        try {
            TwitterFactory twitter = new TwitterFactory();
            this.setScreenName(twitter.getInstance().getScreenName());
            this.setId(twitter.getInstance().getId());
        } catch (TwitterException ex) {
            log.fatal("Exiting. An Exception occurred while establishing twitter client: ", ex);
            System.exit(1);
        }
    }

    @Override
    public void run() {
        TwitterStream twitterStream = new TwitterStreamFactory().getInstance();
        twitterStream.addListener(new TwitterListener());
        twitterStream.filter(getFilterQuery());
        log.info("Start listening to the Twitter stream.");
    }

    private FilterQuery getFilterQuery() {
        FilterQuery filterQuery = new FilterQuery();
        String keywords[] = {this.getTrackedTerm()};
        filterQuery.track(keywords);
        return filterQuery;
    }

    private class TwitterListener implements StatusListener {

        @Override
        public void onStatus(final Status status) {
            log.debug("Received onStatus: " + status.getText());
            synchronized (TwitterClient.this.getMessageData()) {
                TwitterClient.this.getMessageData().addMessage(status.getText());
                TwitterClient.this.getMessageData().notifyAll();
            }
        }

        @Override
        public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {
            log.info("Received a status deletion notice id:" + statusDeletionNotice.getStatusId());
        }

        @Override
        public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
            log.info("Received track limitation notice:" + numberOfLimitedStatuses);
        }

        @Override
        public void onScrubGeo(long userId, long upToStatusId) {
            log.info("Received scrub_geo event userId:" + userId + " upToStatusId:" + upToStatusId);
        }

        @Override
        public void onStallWarning(StallWarning warning) {
            log.info("Received stall warning:" + warning);
        }

        @Override
        public void onException(Exception ex) {
            log.fatal("Received exceptions. Exiting for twitter api safety. onException: ", ex);
            System.exit(1);
        }
    }

}

Lotus.java

package com.gmail.lifeofreilly.lotus;

import org.apache.log4j.Logger;

import java.util.Map;

/**
 * Identifies the top trending hashtags on Twitter for the supplied hashtag, term, or string.
 */
public class Lotus {

    private final static Logger log = Logger.getLogger(Lotus.class);
    private final MessageData messageData;
    private final TwitterClient twitterClient;

    /**
     * Constructs a client using the supplied keyword.
     *
     * @param trackedTerm the term to track on Twitter.
     */
    public Lotus(final String trackedTerm) {
        messageData = new MessageData();
        twitterClient = new TwitterClient(messageData, trackedTerm);
    }

    /**
     * Identifies the top trending hashtags on Twitter for the supplied hashtag, term, or string.
     * Usage: Lotus [keyword, hashtag, or string]
     *
     * @param args required argument. Specifies the keyword or hashtag to track on Twitter.
     */
    public static void main(String[] args) {
        if (args.length == 1) {
            Lotus lotus = new Lotus(args[0]);
            lotus.startTrackingTerm();
            lotus.startProcessingMessages();
            lotus.outputTop10Every30Seconds();
        } else {
            System.out.println("Invalid number of arguments. Usage: Lotus [keyword, hashtag, or string]");
            System.exit(-1);
        }
    }

    /**
     * Get the top 10 hashtags associated with the term being tracked.
     *
     * @return the top 10 hashtags.
     */
    public Map<String, Integer> getTopTenHashtags() {
        return messageData.getTopHashtags(10);
    }

    private void startTrackingTerm() {
        Thread client = new Thread(twitterClient);
        log.info("Starting Twitter client: " + twitterClient.toString() + ".");
        client.start();
    }

    private void startProcessingMessages() {
        MessageProcessor messageProcessor = new MessageProcessor(messageData);
        Thread processor = new Thread(messageProcessor);
        log.info("Starting message processor.");
        processor.start();
    }

    private void outputTop10Every30Seconds() {
        while (true) {
            System.out.println("Top 10 Related Hashtags for the term: " +
                    twitterClient.getTrackedTerm() + ", " +
                    getTopTenHashtags() +
                    ". Total Tweets Processed: " +
                    messageData.getMessageCount());
            try {
                Thread.sleep(30000);
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
            }

        }
    }
}
\$\endgroup\$
6
\$\begingroup\$

Recommendation: use an ExecutorService instead of managing the threads yourself.

You should inject Dependencies, like TwitterClient, into the classes that need them, rather than doing the instantiation locally. If you feel like to absolutely need to hide the implementation detail that is the object graph -- well that's what factories are for.

Similarly, you should inject the dependency you need - if TwitterListener needs to interact with MessageData, then it should be initialized with a reference to it, rather than borrowing the instance from TwitterClient each time. Hard to say - you are inconsistent about whether the MessageData is final (MessageProcessor) or not (AbstractClient). It's not at all clear how this system is supposed to work if everybody isn't sharing the same data.

The TwitterClient constructor is much too busy, and the call to System.exit() is a real shocker. If you were injecting the dependencies you need, you wouldn't need that surprise.

I'm very suspicious of the messageQueue logic, and the locking strategy that you are using. If TwitterClient and MessageProcessor are trying to exchange data via MessageData, then I think all of the synchronization should be inside the MessageData class itself.

Instead of having the timer invoke the getTopTenHashtags(), I'd probably arrange to signal the MessageProcessor thread to invoke it. That protects you from some of the consistency headaches.

You might look at using the Disruptor pattern, instead of trying to work through all of the synchronization issues yourself.

\$\endgroup\$
2
\$\begingroup\$

Generally it looks ok, but there are a few things I don't like.

Do your really need you AbstractClient to contain mutable messageData, trackedTerm, screenName, and id? All of them would look better to be read only fields set by the constructor.

What's the difference between MessageData.messageCount and the value returned by MessageData.messageQueue size() method?

In general, I think that a more even driven approach could be beneficial. What if you have an AbstractClient that allows MessageProcessors to subscribe for new messages? As soon as you have a new message you can feed it to the MessageProcessor, which in turn it will be able to add the relevant information to MessageData. Note if you do that you will have a much simpler MessageProcessor because it will not need anymore to poll MessageData queue.

Basically, the idea is to have a TwitterClient that gets twitter messages, a MessageProcessor that just has the responsibility of analysing them and MessageData to just store the processed data. Does it look any better?

Finally, you decided to run two threads in parallel. That's fine but I think you should pay more attention to concurrency related issues with your queue. It is not synchronised so you could possibly have one thread reading from it and another modifying it, which could lead to crash or to inconsistent results.

\$\endgroup\$
  • \$\begingroup\$ Thanks. I just noticed it and I'm going to review the new version as well \$\endgroup\$ – mariosangiorgio Sep 24 '14 at 20:05

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.