JournalSplitter.java #7

package journal.action;

import jargs.gnu.CmdLineParser;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.PrintStream;
import java.util.zip.Deflater;
import java.util.zip.GZIPOutputStream;

import journal.reader.DataJournalEntry;
import journal.reader.JournalEntry;
import journal.reader.TransactionJournalEntry;

public class JournalSplitter extends BaseAction {

	private String base;
	private String fileName = null;
	private String currentTable = null;
	private boolean compressed = false;
	
	public JournalSplitter() {
		this.base = ".";
	}

	public JournalSplitter(String base) {
		this.base = base;
	}

	@Override
	public void help() {
		System.err.println("\t--action journal.action.JournalSplitter -- -b base");
	}
	
	@Override 
	public String[] parseArgs(String[] args) {

		CmdLineParser parser = new CmdLineParser();
		CmdLineParser.Option baseOption = parser.addStringOption('b', "base");

        try {
            parser.parse(args);
        }
        catch ( CmdLineParser.OptionException e ) {
            System.err.println(e.getMessage());
            help();
            System.exit(2);
        }
        
        this.base = (String) parser.getOptionValue(baseOption, ".");

        return parser.getRemainingArgs();
	}

	@Override
	public void start() throws Exception {
		if (options.isCompressed()) {
			compressed = true;
		}
	}
	
	private void checkFile(DataJournalEntry entry) {
		if (fileName == null) {
			createFile(entry);
		}
		else if ( ! currentTable.equals( entry.getTableName() ) ) {
			closeFile();
		}
	}
	
	private void createFile(DataJournalEntry entry) {
		try {
			currentTable = entry.getTableName();
			fileName = base + File.separatorChar + currentTable + ".ckp";
			if (compressed) fileName += ".gz";
			if( options.isVerbose())
				System.err.println("Next file : " + fileName);
			OutputStream f = new BufferedOutputStream( new FileOutputStream( fileName ) );
			if (compressed) {
				f = new ObjectOutputStream(new GZIPOutputStream(f) {
					{
	                   def.setLevel(Deflater.BEST_COMPRESSION);
	               }});
			}
			out = new PrintStream( f );
		}
		catch( FileNotFoundException e) {
			System.err.println("File not found, probably fileName illegal " + fileName);
			e.printStackTrace(System.err);
			System.exit(1);
		} catch (IOException e) {
			System.err.println("GZIPOutputStream created IO Error");
			e.printStackTrace();
			System.exit(1);
		}
	}
	
	private void closeFile() {
		out.close();
		currentTable = null;
		fileName = null;
	}
	
	public void finish() throws Exception {
		super.finish();
		
		if (fileName != null) {
			// should not happen - unless the last commit is missing
			closeFile(); 
		}
	}
	
	@Override
	public void putValue(DataJournalEntry entry)  throws Exception{
		checkFile((DataJournalEntry) entry);
		out.println(entry.toJournalString());
	}
	
	@Override
	public void replaceValue(DataJournalEntry entry) {
		warnCheckpointOnly();
	}
	
	@Override
	public void deleteValue(DataJournalEntry entry) {
		warnCheckpointOnly();
	}
	
	@Override
	public void verifyValue(DataJournalEntry entry) {
		warnCheckpointOnly();	
	}
	
	private void warnCheckpointOnly() {
		System.err.println("Splitting only makes sense for checkpoint files. Sorry ...");
		System.exit(1);
	}
	
	@Override
	public void commitMarker(TransactionJournalEntry entry) throws Exception {
		// this comes first to write the entry
		// need to write the commit marker to the file before
		// closing it
		
		super.commitMarker(entry);
		
		closeFile();
	}
	
	@Override
	public void flushMarker(TransactionJournalEntry entry) throws Exception {
		super.flushMarker(entry);
	}

}