Merge pull request #22 from didfet/test-bufferedrandomaccessfile

Test bufferedrandomaccessfile + bug fixes

LICENSE.md

@@ -10,4 +10,6 @@ Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
-limitations under the License.
+limitations under the License.
+
+RandomAccessFile and KMPMatch classes by UCAR/Unidata.

pom.xml

@@ -3,7 +3,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>logstash-forwarder-java</groupId>
 	<artifactId>logstash-forwarder-java</artifactId>
-	<version>0.2.4-SNAPSHOT</version>
+	<version>0.2.4-BUFFEREDRANDOMACCESSFILE</version>
 	<name>logstash-forwarder-java</name>
 	<description>Java version of logstash forwarder</description>
 	<url>https://github.com/didfet/logstash-forwarder-java</url>

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

@@ -18,17 +18,19 @@
  */
 
 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;
 import java.util.Map;
 
 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 @@
  *
  */
 
+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;
 
@@ -48,6 +52,8 @@ public class FileState {
 	private FileState oldFileState;
 	@JsonIgnore
 	private Event fields;
+	@JsonIgnore
+	private boolean matchedToNewFile = false;
 
 	public FileState() {
 	}
@@ -56,7 +62,7 @@ public FileState(File file) throws IOException {
 		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();
 	}
@@ -156,6 +162,7 @@ public FileState getOldFileState() {
 
 	public void setOldFileState(FileState oldFileState) {
 		this.oldFileState = oldFileState;
+		oldFileState.setMatchedToNewFile(true);
 	}
 
 	public void deleteOldFileState() {
@@ -172,6 +179,14 @@ public Event getFields() {
 	public void setFields(Event fields) {
 		this.fields = fields;
 	}
+
+	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

@@ -49,12 +49,6 @@ public class FileWatcher {
 	private String sincedbFile = null;
 
 	public FileWatcher() {
-		try {
-			logger.debug("Loading saved states");
-			savedStates = Registrar.readStateFromJson(sincedbFile);
-		} catch(Exception e) {
-			logger.warn("Could not load saved states : " + e.getMessage());
-		}
 	}
 
 	public void initialize() throws IOException {
@@ -194,6 +188,16 @@ private void processModifications() throws IOException {
 				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()))
 				{
@@ -365,7 +369,13 @@ public void setTail(boolean tail) {
 	}
 
 	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/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;
+  }
+}
+