10
\$\begingroup\$

I've been experimenting with the Twitter Streaming API and would like some critical feedback. Specifically code correctness, code smells, overall structure, and my usage of collections and queues.

The application leverages the Twitter Streaming API to identify the top trending hashtags for the supplied hashtag, or string.

Sample invocation:

java -jar ./target/lotus-1.0-SNAPSHOT-jar-with-dependencies.jar apple

Top 10 Hashtags

{#Apple=223, #iTunes=182, #iPhone=160, #Music=62, #Mac=59, #apple=43, #Apps=38, #Movies=25, #iTunesU=21, #Video=19}.

Total Tweets Processed: 1935

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 final String trackedTerm;
    private final MessageData messageData;

    public AbstractClient(final String trackedTerm, final MessageData messageData) {
        this.trackedTerm = trackedTerm;
        this.messageData = messageData;
    }

    public MessageData getMessageData() {
        return messageData;
    }

    public String getTrackedTerm() {
        return trackedTerm;
    }

    @Override
    public String toString() {
        return "AbstractClient{" +
                "trackedTerm='" + trackedTerm + '\'' +
                ", class=" + this.getClass() +
                '}';
    }
}

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.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;

/**
 * A blocking 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 BlockingQueue<String> messageQueue = new LinkedBlockingQueue<String>();
    private volatile 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 total number of messages submitted for processing.
     *
     * @return the number of messages.
     */
    public long getMessageCount() {
        return messageCount;
    }

    /**
     * Removes and returns the head message in the queue, waiting if necessary until an element becomes available.
     *
     * @return the message.
     */
    public String takeMessageFromQueue() {
        String message = "";

        try {
            message = messageQueue.take();
        } catch (InterruptedException ex) {
            log.error("InterruptedException thrown: " + ex);
            Thread.currentThread().interrupt();
        }

        return message;
    }

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

    /**
    * Prints the top ten hashtags to standard out
    */
    public void printTopTenHashTags() {
        System.out.println("Top 10 Hashtags" + getTopHashtags(10) +
            ". Total Tweets Processed: " + getMessageCount());
    }

    /**
     * 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 java.util.StringTokenizer;

/**
 * Extracts hashtags from messages.
 */
public class MessageProcessor implements Runnable {
    private final MessageData messageData;

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

    @Override
    public void run() {
        while (true) {
            extractHashtagsFromMessage(messageData.takeMessageFromQueue());
        }
    }

    private void extractHashtagsFromMessage(final String message) {
        String deliminator = " \t\n\r\f,.:;?![]'";
        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.TwitterStream;
import twitter4j.TwitterStreamFactory;

/**
 * Utilizes the Twitter Streaming API to collect messages.
 */
public 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 trackedTerm the term to track on Twitter.
     * @param messageData the data structure for the Twitter data.
     */
    public TwitterClient(final String trackedTerm, final MessageData messageData) {
        super(trackedTerm, messageData);
    }

    @Override
    public void run() {
        TwitterStream twitterStream = new TwitterStreamFactory().getInstance();
        twitterStream.addListener(new TwitterListener(this.getMessageData()));
        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 {
        private final MessageData messageData;

        public TwitterListener(MessageData messageData) {
            this.messageData = messageData;
        }

        @Override
        public void onStatus(final Status status) {
            log.debug("Received onStatus: " + status.getText());
            messageData.addMessage(status.getText());
        }

        @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.error("Received Exception: ", ex);
        }
    }

}

Lotus.java

package com.gmail.lifeofreilly.lotus;

import org.apache.log4j.Logger;

import twitter4j.Twitter;
import twitter4j.TwitterException;
import twitter4j.TwitterFactory;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 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;
    private final ExecutorService pool = Executors.newFixedThreadPool(2);

    /**
     * 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(trackedTerm, messageData);
    }

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

    private static boolean validCredentialsSupplied() {
        try {
            Twitter twitter = TwitterFactory.getSingleton();
            twitter.verifyCredentials();
            return true;
        } catch (TwitterException ex) {
            System.out.println("Please supply a valid twitter4j.properties file in your working directory. " + ex.getMessage());
            return false;
        }
    }

    private void startTrackingTerm() {
        log.info("Starting Twitter client: " + twitterClient.toString() + ".");
        pool.execute(twitterClient);
    }

    private void startProcessingMessages() {
        log.info("Starting message processor.");
        pool.execute(new MessageProcessor(messageData));
    }

    private void outputTopTenEveryThirtySeconds() {
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            public void run() {
                messageData.printTopTenHashTags();
            }
        }, 0, 30000);
    }
}
\$\endgroup\$
2
  • \$\begingroup\$ It smells of race condition (I might be wrong) of messageCount++;. Just because it's volatile doesn't mean it wont be tried to be updated in two threads simultaneously; Especially being a long and all. I don't know multi-threading, but you should maybe surround with synchronized, or add synchronized to the method. \$\endgroup\$
    – Pål GD
    Sep 23 '14 at 19:18
  • \$\begingroup\$ Hi, re: messageCount++ It's just a simple counter, it does not control flow. Based on my understanding of this keyword it seems to be fine. 'Access to the variable acts as though it is enclosed in a synchronized block, synchronized on itself.' javamex.com/tutorials/synchronization_volatile.shtml \$\endgroup\$
    – weare138
    Sep 24 '14 at 20:16
6
+50
\$\begingroup\$

The following comment is a very bad sign:

/**
 * A blocking message queue and the hashtags extracted.
 */
public class MessageData {

Right there in the JavaDoc you are telling the world that MessageData is filling two completely different roles. You should separate the two roles out - make it explicitly clear that the queue of work to be done, and the report of the work done thus far, are two different things.

Another way of saying the same thing: your pipeline has two different stages in it. Stage one reads messages and writes hashtags. Stage two reads hashtags and writes updates to an in memory database. Create classes to manage each of those responsibilities.

Currently the hashtags are stored in memory, but in a subsequent version I plan to move to mongo

Yeah - that right there is a big hint that you are going to want to be able to swap out different implementations of your hash tag store.

You'll know that you have the right design when you can create a single-threaded unit test that is able to verify the flow of a message all the way through the processing.

@Override
public void run() {
    while (true) {
        extractHashtagsFromMessage(messageData.takeMessageFromQueue());
    }
}

public String takeMessageFromQueue() {
    String message = "";

    try {
        message = messageQueue.take();
    } catch (InterruptedException ex) {
        log.error("InterruptedException thrown: " + ex);
        Thread.currentThread().interrupt();
    }

    return message;
}

These two functions show that you don't understand what the InterruptedException is for.

The InterruptedException wasn't left in the method signature by mistake; it signals a condition that correctly written programs should be prepared to handle -- namely, that some other thread has discovered that the blocking process should be cancelled.

The quick fix would be to simply handle the interrupted condition in the Runnable.

public void run() {
    while (! Thread.interrupted()) {
        extractHashtagsFromMessage(messageData.takeMessageFromQueue());
    }
} 

But it's still a little bit weird that MessageData returns a fake message during cancellation. That would mean that MessageProcessor is consuming more messages than MessageData.getMessageCount() claims were put in the queue.

There are two ways you might fix that. One would be to move the InterruptedException to the throws clause, and let the Runnable handle it. Another option would be to modify the signature of takeMessageFromQueue so that it accepts a Listener.

public void takeMessageFromQueue(Listener listener) {
    try {
        String message = messageQueue.take();
        listener.onMessage(message);
    } catch (InterruptedException ex) {
        log.error("InterruptedException thrown: " + ex);
        Thread.currentThread().interrupt();
    }
}

This approach has the additional advantage that it allows you to experiment with other strategies for handling a backlog of messages; you could drain the entire queue in one go, or process a batch of messages (which allows you to take advantage of other data structures that are optimized for that use case).

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

Ick - use Dependency Injection. It's ever so much friendlier to build the object graph explicitly, right where everyone is looking for it. If you must create a new graph each time run() is called, well that's what Factories are for....

I'm alarmed that you seem to be dismissing the big chicken in the yard. Your message source is Twitter; you're drinking from the fire hose, but you don't seem to have given yourself a way to purge the data being shoved down your throat.

private final Multiset<String> hashTags = TreeMultiset.create();

How many different tags do you think you are going to see exactly once? They are useless for your report, but your memory is going to be flooded with them.

You can probably get away with the heuristic that any tag that hasn't appeared in some reasonable time interval isn't going to be a top 10 tag. Simplest tool in the box would likely be a cache with a reasonable eviction policy built into it.

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;
}

Ow. Look carefully at this code -- you are going to sort every hash tag you have in memory, and then crop away only the top N? That's a lot of wear and tear on your CPU for a very small result.

Now, if your requirement really is that you have to provide the top 1 billion hash tags on request, then you may be stuck. But if the real number you need is in the ballpark of 10 or 100, then you should keep a running count going in memory, and hand out the latest snapshot as needed (which can be trimmed down if the caller doesn't need as many items as you provide).

A PriorityQueue gets you a lot of the way there - it's a data structure that can tell you the minimum element in it in O(1) time. A simple approach would be to scan all of your tags, and if the tag is larger than the current minimum, then remove the old minimum and offer the new tag -- the data structure knows how to order the new value correctly.

Of course, scanning all of the tags each time somebody asks for the top 10 list may still be too expensive. You might prefer to pre-calculate the top tags. That's a fine idea, but you have to be a careful when the priority of an item already in the queue changes. You can manage that by removing and re-inserting each object in the queue that updates. The cost of that operation for queue size N is O(N), which shouldn't be a problem when N is 10 or 100. You can get O(log(N)) performance if you feel up to writing your own Heap

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

This implementation looks better than the one in the original question. As I said in my previous answer, your design looks a bit too complex.

I'd go for a more reactive, event based solution. I'd structure your client and your message processor in the following way.

interface Client
{
    void registerMessageProcessor(MessageProcessor messageProcessor);
}

interface MessageProcessor
{
    void onNewMessages(Collection<Message> messages);
}

The core idea is to have a client that gets data from the twitter stream, allows some one or more message processors to subscribe for events and notifies them when there are new messages available. These interface are completely agnostic with respect to what you need to do with your messages.

Let's address the specific problem you want to tackle: processing the messages, analysing them and storing them in a data structure. You only need to query the hashtags and you don't really care about the messages, so I'd avoid storing them. Let's use a simple hashtag counter. Again I'm just showing the interface. To address possible threading issue you should consider using a thread safe multi set.

interface HashTagCounter
{
    void addHashTag(String hashTag);
    Map<String,Integer> topHashTags(int maxNumberOfHashTags);
}

Your message processors onNewMessages method will just need to find the hash tags in the messages and call addHashTag.

Finally, you need to have a separate Reporter thread that polls the HashTagCounter periodically to output the top hash tags. That should be simple enough as it should just call topHashTags and nicely format the result.

In your main you have to wire up all your dependencies in the following way.

public static void main(String[] args)
{
    Client client = new TwitterClient(/* the args you need */);

    HashTagCounter hashTagCounter = new HashTagCounter();
    Reporter reporter = new Reporter(hashTagCounter);
    String keyword = "Code review";
    MessageProcessor messageProcessor = new MessageProcessor(keyword, hashTagCounter)

    client.registerMessageProcessor(messageProcessor);

    client.start();
    reporter.start();        
}

Note that the MessageProcessor runs on the TwitterClient thread.

This should work fine in a basic use case. As they suggested in your previous question, if you are looking for high performance and you need to address more complex scenarios you should look for a disruptor-like approach.

\$\endgroup\$
2
  • \$\begingroup\$ Thanks again for your feedback Mario. I'm just now working my way through Head First Design Patterns by Eric Freeman. I like the idea of trying out the Observer pattern for this use case as you suggested. I have yet to read up on disruptors but they look very interesting. Cheers. ref: martinfowler.com/articles/lmax.html \$\endgroup\$
    – weare138
    Sep 26 '14 at 17:27
  • \$\begingroup\$ If you like the observer pattern and want to create something more complex, combining event sources in different ways have a look at RxJava. I think it is a very interesting approach \$\endgroup\$ Sep 26 '14 at 17:33

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.