src/main/java/org/lwes/journaller/DeJournaller.java

   1 package org.lwes.journaller;
   2 /**
   3  * A Class that will "dejournal" files written to by the GZIPEventHandler.  If you want to
   4  * do something other than write the events to stdout, subclass this and override handleEvent.
   5  *
   6  * User: fmaritato
   7  * Date: Apr 16, 2009
   8  */
   9
  10 import org.apache.commons.logging.Log;
  11 import org.apache.commons.logging.LogFactory;
  12 import org.apache.hadoop.conf.Configuration;
  13 import org.apache.hadoop.fs.FileSystem;
  14 import org.apache.hadoop.fs.Path;
  15 import org.apache.hadoop.io.BytesWritable;
  16 import org.apache.hadoop.io.SequenceFile;
  17 import org.kohsuke.args4j.CmdLineException;
  18 import org.kohsuke.args4j.CmdLineParser;
  19 import org.kohsuke.args4j.Option;
  20 import org.lwes.Event;
  21 import org.lwes.EventSystemException;
  22 import org.lwes.db.EventTemplateDB;
  23 import org.lwes.journaller.util.EventHandlerUtil;
  24 import org.lwes.serializer.DeserializerState;
  25
  26 import java.io.ByteArrayInputStream;
  27 import java.io.DataInputStream;
  28 import java.io.EOFException;
  29 import java.io.File;
  30 import java.io.FileInputStream;
  31 import java.io.IOException;
  32 import java.util.zip.GZIPInputStream;
  33
  34 public class DeJournaller implements Runnable, JournallerConstants {
  35     private static transient Log log = LogFactory.getLog(DeJournaller.class);
  36
  37     @Option(name = "--seq", aliases = "--sequence")
  38     protected boolean sequence = false;
  39
  40     @Option(name = "-g", aliases = "--gzipped")
  41     protected boolean gzipped;
  42
  43     @Option(name = "-f", aliases = "--file")
  44     protected String fileName;
  45
  46     @Option(name = "-e", aliases = "--esf-file")
  47     protected String esfFile;
  48
  49     @Option(name = "-v", aliases = "validate")
  50     protected boolean validate = false;
  51
  52     public DeJournaller() {
  53     }
  54
  55     public void run() {
  56
  57         if (fileName == null || "".equals(fileName)) {
  58             log.error("File name was not specified");
  59             return;
  60         }
  61
  62         EventTemplateDB evtTemplate = new EventTemplateDB();
  63         DeserializerState state = new DeserializerState();
  64
  65         if (getEsfFile() != null) {
  66             evtTemplate.setESFFile(new File(getEsfFile()));
  67         }
  68         log.debug("esf: " + evtTemplate.getESFFile());
  69         log.debug("validate: " + validate);
  70         evtTemplate.initialize();
  71
  72         DataInputStream in = null;
  73         try {
  74             log.debug("Opening file: " + fileName);
  75             if (sequence) {
  76                 processSequenceFile(fileName);
  77             }
  78             else {
  79                 if (gzipped) {
  80                     in = new DataInputStream(new GZIPInputStream(new FileInputStream(fileName)));
  81                 }
  82                 else {
  83                     in = new DataInputStream(new FileInputStream(fileName));
  84                 }
  85                 Event evt;
  86                 while ((evt = EventHandlerUtil.readEvent(in, state, evtTemplate, validate)) != null) {
  87                     state.reset();
  88                     handleEvent(evt);
  89                 }
  90             }
  91         }
  92         catch (EOFException e) {
  93             // this is normal. Just catch and ignore.
  94         }
  95         catch (Exception e) {
  96             log.error(e.getMessage(), e);
  97         }
  98         finally {
  99             if (in != null) {
 100                 try {
 101                     in.close();
 102                 }
 103                 catch (IOException e) {
 104                     log.error(e.getMessage(), e);
 105                 }
 106             }
 107             done();
 108         }
 109     }
 110
 111     protected void processSequenceFile(String file) {
 112         Configuration conf = new Configuration();
 113         FileSystem fs = null;
 114         SequenceFile.Reader reader = null;
 115         try {
 116             if (log.isDebugEnabled()) {
 117                 log.debug("Opening file: " + file);
 118             }
 119             fs = FileSystem.get(conf);
 120             Path p = new Path(file);
 121             reader = new SequenceFile.Reader(fs, p, conf);
 122
 123             BytesWritable key = (BytesWritable) reader.getKeyClass().newInstance();
 124             BytesWritable value = (BytesWritable) reader.getKeyClass().newInstance();
 125
 126             EventTemplateDB templ = new EventTemplateDB();
 127             DeserializerState state = new DeserializerState();
 128             while (reader.next(key, value)) {
 129                 byte[] evtBytes = new byte[key.getLength()+value.getLength()];
 130                 System.arraycopy(key.getBytes(), 0,
 131                                  evtBytes, 0,
 132                                  key.getLength());
 133                 System.arraycopy(value.getBytes(), 0,
 134                                  evtBytes,
 135                                  key.getLength(), value.getLength());
 136                 Event evt = EventHandlerUtil.readEvent
 137                     (new DataInputStream(new ByteArrayInputStream(evtBytes)),
 138                      state, templ);
 139                 state.reset();
 140                 if (log.isDebugEnabled()) {
 141                     log.debug("read a k/v: "+evt.toOneLineString());
 142                 }
 143                 handleEvent(evt);
 144             }
 145         }
 146         catch (IOException e) {
 147             log.error(e.getMessage(), e);
 148         }
 149         catch (InstantiationException e) {
 150             log.error(e.getMessage(), e);
 151         }
 152         catch (IllegalAccessException e) {
 153             log.error(e.getMessage(), e);
 154         }
 155         catch (EventSystemException e) {
 156             log.error(e.getMessage(), e);
 157         }
 158         finally {
 159             if (reader != null) {
 160                 try {
 161                     reader.close();
 162                 }
 163                 catch (IOException e) {
 164                     log.error(e.getMessage(), e);
 165                 }
 166             }
 167         }
 168     }
 169
 170     /**
 171      * This method is called after all events in the file have been processed.
 172      */
 173     public void done() {
 174         if (log.isDebugEnabled()) {
 175             log.debug("done");
 176         }
 177     }
 178
 179     /**
 180      * Subclass and override this method if you want to do something other than print the
 181      * events to the logging system. For example, you may want to write these events to a
 182      * database.
 183      *
 184      * @param event
 185      */
 186     public void handleEvent(Event event) {
 187         System.out.println(event.toOneLineString());
 188     }
 189
 190     protected void parseArguments(String[] args) throws CmdLineException {
 191         CmdLineParser parser = new CmdLineParser(this);
 192         parser.parseArgument(args);
 193     }
 194
 195     public static void main(String[] args) {
 196         DeJournaller dj = new DeJournaller();
 197         try {
 198             dj.parseArguments(args);
 199         }
 200         catch (CmdLineException e) {
 201             log.error(e.getMessage(), e);
 202         }
 203         dj.run();
 204     }
 205
 206     public String getFileName() {
 207         return fileName;
 208     }
 209
 210     public void setFileName(String fileName) {
 211         this.fileName = fileName;
 212     }
 213
 214     public String getEsfFile() {
 215         return esfFile;
 216     }
 217
 218     public void setEsfFile(String esfFile) {
 219         this.esfFile = esfFile;
 220     }
 221
 222     public boolean isGzipped() {
 223         return gzipped;
 224     }
 225
 226     public void setGzipped(boolean gzipped) {
 227         this.gzipped = gzipped;
 228     }
 229
 230     public boolean isValidate() {
 231         return validate;
 232     }
 233
 234     public void setValidate(boolean validate) {
 235         this.validate = validate;
 236     }
 237 }