merged Multiline capability

src/main/java/info/fetter/logstashforwarder/FileReader.java

@@ -18,10 +18,11 @@ package info.fetter.logstashforwarder;
  */
 
 import info.fetter.logstashforwarder.util.AdapterException;
+import info.fetter.logstashforwarder.util.RandomAccessFile;
 
 import java.io.File;
 import java.io.IOException;
-import java.io.RandomAccessFile;
+//import java.io.RandomAccessFile;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
@@ -30,6 +31,7 @@ import org.apache.commons.lang.ArrayUtils;
 
 import org.apache.log4j.Logger;
 
+
 public class FileReader extends Reader {
 	private static Logger logger = Logger.getLogger(FileReader.class);
 	private static final byte[] ZIP_MAGIC = new byte[] {(byte) 0x50, (byte) 0x4b, (byte) 0x03, (byte) 0x04};

src/main/java/info/fetter/logstashforwarder/FileSigner.java

@@ -1,9 +1,12 @@
 package info.fetter.logstashforwarder;
 
+import info.fetter.logstashforwarder.util.RandomAccessFile;
+
 import java.io.IOException;
-import java.io.RandomAccessFile;
+//import java.io.RandomAccessFile;
 import java.util.zip.Adler32;
 
+
 public class FileSigner {
 	private static final Adler32 adler32 = new Adler32();
 	

src/main/java/info/fetter/logstashforwarder/FileState.java

@@ -17,10 +17,14 @@ package info.fetter.logstashforwarder;
  *
  */
 
+import info.fetter.logstashforwarder.util.RandomAccessFile;
+
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
-import java.io.RandomAccessFile;
+//import java.io.RandomAccessFile;
+
+
 
 import org.apache.commons.lang.builder.ToStringBuilder;
 
@@ -49,10 +53,10 @@ public class FileState {
 	private FileState oldFileState;
 	@JsonIgnore
 	private Event fields;
-        @JsonIgnore
-        private Multiline multiline;
-        @JsonIgnore
-        private byte[] bufferedLines = null;
+    @JsonIgnore
+    private Multiline multiline;
+	@JsonIgnore
+	private boolean matchedToNewFile = false;
 
 	public FileState() {
 	}
@@ -61,7 +65,7 @@ public class FileState {
 		this.file = file;
 		directory = file.getCanonicalFile().getParent();
 		fileName = file.getName();
-		randomAccessFile = new RandomAccessFile(file, "r");
+		randomAccessFile = new RandomAccessFile(file.getPath(), "r");
 		lastModified = file.lastModified();
 		size = file.length();
 	}
@@ -161,6 +165,7 @@ public class FileState {
 
 	public void setOldFileState(FileState oldFileState) {
 		this.oldFileState = oldFileState;
+		oldFileState.setMatchedToNewFile(true);
 	}
 	
 	public void deleteOldFileState() {
@@ -177,22 +182,23 @@ public class FileState {
 	public void setFields(Event fields) {
 		this.fields = fields;
 	}
+	
+    public Multiline getMultiline() {
+           return multiline;
+    }
+    
+    public void setMultiline(Multiline multiline) {
+           this.multiline = multiline;
+    }
         
-        public Multiline getMultiline() {
-               return multiline;
-        }
-        
-        public void setMultiline(Multiline multiline) {
-               this.multiline = multiline;
-        }
-        
-        public byte[] getBufferedLines() {
-               return bufferedLines;
-        }
-               
-        public void setBufferedLines(byte[] bufferedLines) {
-               this.bufferedLines = bufferedLines;
-        }
+	
+	public boolean isMatchedToNewFile() {
+		return matchedToNewFile;
+	}
+
+	public void setMatchedToNewFile(boolean matchedToNewFile) {
+		this.matchedToNewFile = matchedToNewFile;
+	}
 
 	@Override
 	public String toString() {

src/main/java/info/fetter/logstashforwarder/FileWatcher.java

@@ -48,14 +48,7 @@ public class FileWatcher {
 	private boolean stdinConfigured = false;
 	private String sincedbFile = null;
 
-	public FileWatcher(String sincedbFileName) {
-		sincedbFile = sincedbFileName;
-		try {
-			logger.debug("Loading saved states");
-			savedStates = Registrar.readStateFromJson(sincedbFile);
-		} catch(Exception e) {
-			logger.warn("Could not load saved states : " + e.getMessage());
-		}
+	public FileWatcher() {
 	}
 
 	public void initialize() throws IOException {
@@ -195,6 +188,16 @@ public class FileWatcher {
 				if(logger.isDebugEnabled()) {
 					logger.debug("File " + state.getFile() + " has been truncated or created, not retrieving pointer");
 				}
+				oldState = oldWatchMap.get(state.getFile());
+				if(oldState != null && ! oldState.isMatchedToNewFile()) {
+					if(logger.isDebugEnabled()) {
+						logger.debug("File " + state.getFile() + " has been replaced and not renamed, removing from watchMap");
+					}
+					try {
+						oldState.getRandomAccessFile().close();
+					} catch(Exception e) {}
+					oldWatchMap.remove(state.getFile());
+				}
 			} else {
 				if(logger.isInfoEnabled() && ! state.getFileName().equals(oldState.getFileName()))
 				{
@@ -367,6 +370,13 @@ public class FileWatcher {
 	}
 
 	public void setSincedb(String sincedbFile) {
-		this.sincedbFile = sincedbFile;	
-        }
+		this.sincedbFile = sincedbFile;
+		try {
+			logger.debug("Loading saved states");
+			savedStates = Registrar.readStateFromJson(sincedbFile);
+		} catch(Exception e) {
+			logger.warn("Could not load saved states : " + e.getMessage(), e);
+		}
+	}
+
 }

src/main/java/info/fetter/logstashforwarder/Forwarder.java

@@ -72,9 +72,10 @@ public class Forwarder {
 		try {
 			parseOptions(args);
 			setupLogging();
-			watcher = new FileWatcher(sincedbFile);
+			watcher = new FileWatcher();
 			watcher.setMaxSignatureLength(signatureLength);
 			watcher.setTail(tailSelected);
+			watcher.setSincedb(sincedbFile);
 			configManager = new ConfigurationManager(config);
 			configManager.readConfiguration();
 			for(FilesSection files : configManager.getConfig().getFiles()) {

src/main/java/info/fetter/logstashforwarder/util/KMPMatch.java

@@ -0,0 +1,127 @@
+/*
+ * Copyright 1998-2009 University Corporation for Atmospheric Research/Unidata
+ *
+ * Portions of this software were developed by the Unidata Program at the
+ * University Corporation for Atmospheric Research.
+ *
+ * Access and use of this software shall impose the following obligations
+ * and understandings on the user. The user is granted the right, without
+ * any fee or cost, to use, copy, modify, alter, enhance and distribute
+ * this software, and any derivative works thereof, and its supporting
+ * documentation for any purpose whatsoever, provided that this entire
+ * notice appears in all copies of the software, derivative works and
+ * supporting documentation.  Further, UCAR requests that the user credit
+ * UCAR/Unidata in any publications that result from the use of this
+ * software or in any product that includes this software. The names UCAR
+ * and/or Unidata, however, may not be used in any advertising or publicity
+ * to endorse or promote any products or commercial entity unless specific
+ * written permission is obtained from UCAR/Unidata. The user also
+ * understands that UCAR/Unidata is not obligated to provide the user with
+ * any support, consulting, training or assistance of any kind with regard
+ * to the use, operation and performance of this software nor to provide
+ * the user with any updates, revisions, new versions or "bug fixes."
+ *
+ * THIS SOFTWARE IS PROVIDED BY UCAR/UNIDATA "AS IS" AND ANY EXPRESS OR
+ * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL UCAR/UNIDATA BE LIABLE FOR ANY SPECIAL,
+ * INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING
+ * FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
+ * NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
+ * WITH THE ACCESS, USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+package info.fetter.logstashforwarder.util;
+
+/**
+ * Knuth-Morris-Pratt Algorithm for Pattern Matching.
+ * Immutable
+ *
+ * @author caron
+ * @see <a href="http://www.fmi.uni-sofia.bg/fmi/logic/vboutchkova/sources/KMPMatch_java.html">http://www.fmi.uni-sofia.bg/fmi/logic/vboutchkova/sources/KMPMatch_java.html</a>
+ * @since May 9, 2008
+ */
+public class KMPMatch {
+
+  private final byte[] match;
+  private final int[] failure;
+
+  /**
+   * Constructor
+   * @param match search for this byte pattern
+   */
+  public KMPMatch(byte[] match) {
+    this.match = match;
+    failure = computeFailure(match);
+  }
+
+  public int getMatchLength() { return match.length; }
+
+  /**
+   * Finds the first occurrence of match in data.
+   * @param data search in this byte block
+   * @param start start at data[start]
+   * @param max end at data[start+max]
+   * @return index into data[] of first match, else -1 if not found.
+   */
+  public int indexOf(byte[] data, int start, int max) {
+    int j = 0;
+    if (data.length == 0) return -1;
+
+    for (int i = start; i < start + max; i++) {
+      while (j > 0 && match[j] != data[i])
+        j = failure[j - 1];
+
+      if (match[j] == data[i])
+        j++;
+
+      if (j == match.length)
+        return i - match.length + 1;
+
+    }
+    return -1;
+  }
+
+  /*
+   * Finds the first occurrence of match in data.
+   * @param data search in this byte block
+   * @param start start at data[start]
+   * @param max end at data[start+max]
+   * @return index into block of first match, else -1 if not found.
+   *
+  public int scan(InputStream is, int start, int max) {
+    int j = 0;
+    if (data.length == 0) return -1;
+
+    for (int i = start; i < start + max; i++) {
+      while (j > 0 && match[j] != data[i])
+        j = failure[j - 1];
+
+      if (match[j] == data[i])
+        j++;
+
+      if (j == match.length)
+        return i - match.length + 1;
+
+    }
+    return -1;
+  } // */
+
+
+  private int[] computeFailure(byte[] match) {
+    int[] result = new int[match.length];
+
+    int j = 0;
+    for (int i = 1; i < match.length; i++) {
+      while (j > 0 && match[j] != match[i])
+        j = result[j - 1];
+
+      if (match[i] == match[i])
+        j++;
+
+      result[i] = j;
+    }
+
+    return result;
+  }
+}
+

src/test/java/info/fetter/logstashforwarder/FileReaderTest.java

@@ -56,7 +56,7 @@ public class FileReaderTest {
 		List<FileState> fileList = new ArrayList<FileState>(1);
 		File file1 = new File("testFileReader1.txt");
 		FileUtils.write(file1, "testFileReader1 line1\n");
-                FileUtils.write(file1, " nl line12\n", true);
+        FileUtils.write(file1, " nl line12\n", true);
 		FileUtils.write(file1, "testFileReader1 line2\n", true);
 		FileUtils.write(file1, "testFileReader1 line3\n", true);
 		Thread.sleep(500);

src/test/java/info/fetter/logstashforwarder/FileWatcherTest.java

@@ -48,7 +48,7 @@ public class FileWatcherTest {
 
 	//@Test
 	public void testFileWatch() throws InterruptedException, IOException {
-		FileWatcher watcher = new FileWatcher(".logstash-forwarder-java");
+		FileWatcher watcher = new FileWatcher();
 		watcher.addFilesToWatch("./test.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY, null);
 		for(int i = 0; i < 100; i++) {
 			Thread.sleep(1000);
@@ -58,7 +58,7 @@ public class FileWatcherTest {
         
         //@Test
         public void testFileWatchWithMultilines() throws InterruptedException, IOException {
-		FileWatcher watcher = new FileWatcher(".logstash-forwarder-java");
+		FileWatcher watcher = new FileWatcher();
                 Multiline multiline = new Multiline();
 		watcher.addFilesToWatch("./test.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY, multiline);
 		for(int i = 0; i < 100; i++) {
@@ -73,7 +73,7 @@ public class FileWatcherTest {
 			logger.warn("Not executing this test on windows");
 			return;
 		}
-		FileWatcher watcher = new FileWatcher(".logstash-forwarder-java");
+		FileWatcher watcher = new FileWatcher();
 		watcher.addFilesToWatch("./testFileWatcher*.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY, null);
 		watcher.initialize();
 
@@ -118,7 +118,7 @@ public class FileWatcherTest {
 			logger.warn("Not executing this test on windows");
 			return;
 		}
-		FileWatcher watcher = new FileWatcher(".logstash-forwarder-java");
+		FileWatcher watcher = new FileWatcher();
                 Map<String, String> m = new HashMap<String, String>();
                 m.put("pattern", " nl");
                 m.put("negate", "false");