path: root/src/main/Main.java

package main;

import com.google.gson.JsonParseException;
import com.google.gson.JsonSyntaxException;
import data.Tweet;
import database.ConnectionBuilder;
import io.FileTweetReader;
import io.ITweetReader;
import io.TweetReader;
import java.io.IOException;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * The main class.
 */
public class Main implements Callable<Boolean> {

    /**
     * The main method of the application.
     *
     * @param args the global arguments to pass to the program.
     */
    public static void main(String[] args) {
        Main main;
        try {
            main = new Main(args);
        } catch (IllegalArgumentException ex) {
            System.err.println(ex.getMessage());
            System.exit(1);
            return;
        }

        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
        // the main IO thread
        Future<Boolean> mainTask = scheduler.submit(main);

        // the status thread
        if (main.statusInterval > 0) {
            scheduler.scheduleAtFixedRate(new Watcher(main, main.statusInterval),
                    main.statusInterval, main.statusInterval, TimeUnit.SECONDS);
        }
        // now wait for completion of the main thread
        try {
            while (true) {
                try {
                    if (mainTask.get()) {
                        System.out.println("Import successful.");
                    } else {
                        System.out.println("Import (partially) failed.");
                    }
                    break;
                } catch (InterruptedException ex) {
                    Logger.getLogger(Main.class.getName())
                            .info("Interrupted while waiting for main");
                }
            }
        } catch (ExecutionException ex) {
            Logger.getLogger(Main.class.getName())
                    .log(Level.SEVERE, "Main failed", ex);
        }
        // cancel status thread
        scheduler.shutdown();
    }

    private String m_filename;
    private final ConnectionBuilder cb;
    /**
     * Whether the database should be contacted or not.
     */
    private boolean skipDb;
    private Integer statusInterval;

    public Main(String[] args) {
        // default connection properties
        cb = new ConnectionBuilder()
                .setServerName("localhost")
                .setUsername("twitter")
                .setPassword("2IOC02")
                .setDbName("twitter");
        skipDb = false;
        statusInterval = 2;

        /* parse the global options. */
        parseGlobalOptions(args);
    }

    /**
     * The current tweet number that is being processed.
     */
    private volatile int tweetNo;

    private boolean printTweets(ITweetReader reader) throws IOException {
        Tweet tweet;
        tweetNo = 1;
        while ((tweet = reader.getTweet()) != null) {
            System.out.println("/*" + tweetNo++ + "*/ " + tweet);
        }
        return true;
    }

    private boolean tweetsToDb(ITweetReader reader) throws IOException {
        Tweet tweet = null;
        tweetNo = 1;
        try (Connection connection = cb.create()) {
            /* create the object that fills the database */
            DataFiller filler = new DataFiller(connection);
            while ((tweet = reader.getTweet()) != null) {
                filler.processTweet(tweet);
                ++tweetNo;
            }
            return true;
        } catch (JsonParseException ex) {
            if (tweet != null) {
                System.err.println("Faulty tweet " + tweetNo + ": " + tweet);
            }
            Logger.getLogger(Main.class.getName()).log(Level.SEVERE,
                    "Tweet read error", ex);
        } catch (SQLException ex) {
            if (tweet != null) {
                System.err.println("Faulty tweet " + tweetNo + ": " + tweet);
            }
            Logger.getLogger(Main.class.getName()).log(Level.SEVERE,
                    "DB error", ex);
        }
        return false;
    }

    public int getProcessedTweets() {
        return tweetNo;
    }

    @Override
    public Boolean call() {
        ITweetReader reader = null;
        boolean success = false;
        try {
            if (m_filename == null) {
                reader = new TweetReader(System.in);
            } else {
                reader = new FileTweetReader(m_filename);
            }
            if (skipDb) {
                success = printTweets(reader);
            } else {
                success = tweetsToDb(reader);
            }
        } catch (JsonSyntaxException ex) {
            System.err.println("Got an invalid tweet: " + ex);
        } catch (IOException ex) {
            System.err.println("Cannot open tweets: " + ex);
        } finally {
            if (reader != null) {
                reader.close();
            }
        }

        return success;
    }

    private void parseGlobalOptions(String[] args)
            throws IllegalArgumentException {
        /* parse global options */
        for (int i = 0; i < args.length; i++) {
            if ("--help".equals(args[i])) {
                printHelp();
                System.exit(0);
            } else if ("--dbhost".equals(args[i])) {
                cb.setServerName(getArg(args, ++i, "--dbhost"));
            } else if ("--dbuser".equals(args[i])) {
                cb.setUsername(getArg(args, ++i, "--dbuser"));
            } else if ("--dbpass".equals(args[i])) {
                cb.setPassword(getArg(args, ++i, "--dbpass"));
            } else if ("--dbport".equals(args[i])) {
                cb.setPort(Integer.valueOf(getArg(args, ++i, "--dbport")));
            } else if ("--dbname".equals(args[i])) {
                cb.setDbName(getArg(args, ++i, "--dbname"));
            } else if ("--skipdb".equals(args[i])) {
                skipDb = true;
            } else if ("--status".equals(args[i])) {
                statusInterval = Integer.valueOf(getArg(args, ++i, "--status"));
            } else if (args[i].startsWith("-")) {
                throw new IllegalArgumentException("Invalid option: " + args[i]);
            } else {
                /* This should be the filename */
                m_filename = args[i];
            }
        }
    }

    /**
     * Read an extra option for a command.
     */
    private String getArg(String[] params, int index, String name) {
        if (index >= params.length) {
            System.err.println("Missing argument for parameter " + name);
            System.exit(1);
        }
        return params[index];
    }

    /**
     * Print some useful help messages.
     */
    private void printHelp() {
        for (String line : HELP) {
            System.out.println(line);
        }
    }

    private final static String[] HELP = {
        "Usage: java -jar DataFiller.jar [options] [tweets-file]",
        "",
        "Global options:",
        "   --help        Print this help text.",
        "   --dbhost HOST Database host (defaults to 'localhost')",
        "   --dbuser USER Database username (defaults to 'postgres')",
        "   --dbpass PASS Database password (defaults to '2IOC02')",
        "   --dbport PORT Database port (defaults to 5432)",
        "   --dbname NAME Database name (defaults to 'Twitter')",
        "   --skipdb      Do not contact the database at all, just print data.",
        "   --status SECS The interval in which import status information",
        "                 should be printed, zero disables it (defaults to 2)",
        "",
        "If no tweets file is given, data will be read from standard input."
    };
}