fix duplicate content when soft link, replace lumberjack to tcp protocol

.gitignore

@@ -3,4 +3,6 @@
 /testFileReader1.txt
 /state2.json
 .idea/
-logstash-forwarder-java.iml
+logstash-forwarder-java.iml
+.DS_Store
+.logstash-forwarder-java

README.md

@@ -1,62 +1,73 @@
-# logstash-forwarder-java
+# logstash-forwarder-java(tcp-input)
 
-## What is this ?
+Forked from https://github.com/didfet/logstash-forwarder-java
 
-Logstash-forwarder-java is a log shipper program written in java. This is in fact a java version of [logstash-forwarder](https://github.com/elasticsearch/logstash-forwarder) by jordansissel.
-Here are a few features of this program :
-  - compatible with Java 5 runtime
-  - lightweight : package size is ~2MB and memory footprint ~8MB
-  - configuration compatible with logstash-forwarder
-  - lumberjack output (including zlib compression)
+Logstash Lumberjack input is not robust enough, this fork replaced Lumberjack protocol to plain TCP.
 
-## Why ?
+## Example
 
-Logstash-forwarder is written in go. This programming language is not available on all platforms (for example AIX), that's why a java version is more portable.
+### Logstash pipelines
 
-Logstash runs on java and provides a lumberjack output, but the file input doesn't run on all plaforms (for example AIX) and logstash requires a recent JVM. Moreover Logstash is heavier : it is a big package and uses more system resources.
+```
+input {
+    tcp {
+        port => 5055
+        codec => json_lines
+        #ssl_cert => "/etc/logstash/server.crt"
+        #ssl_certificate_authorities => "/etc/logstash/ca.crt"
+        #ssl_enable => true
+        #ssl_key => "/etc/logstash/server.key"
+        #ssl_key_passphrase => "${TCP_KEY_PASS}"
+        #ssl_verify => false
+        tcp_keep_alive => true
+    }
+}
 
-So logstash-forwarder-java is a solution for those who want a portable, lightweight log shipper for their ELK stack.
+filter {
+}
 
-## How to install it ?
+output {
+   stdout{} 
+}
+```
+### Run logstash-forwarder
 
-Download one of the following archives :
-  - [logstash-forwarder-java-0.2.4-bin.zip](https://github.com/didfet/logstash-forwarder-java/releases/download/0.2.4/logstash-forwarder-java-0.2.4-bin.zip)
-  - [logstash-forwarder-java-0.2.4-bin.tar.gz](https://github.com/didfet/logstash-forwarder-java/releases/download/0.2.4/logstash-forwarder-java-0.2.4-bin.tar.gz)
-  - [logstash-forwarder-java-0.2.4-bin.tar.bz2](https://github.com/didfet/logstash-forwarder-java/releases/download/0.2.4/logstash-forwarder-java-0.2.4-bin.tar.bz2)
+Just run this command :
 
-Or download the maven project and run maven package. Then you can install one of the archives located in the target directory.
+    java -jar logstash-forwarder-java-X.Y.Z.jar -config /path/to/config/file.json
 
-## How to run it ?
+### Run logstash-forwarder with embedded JRE
 
-Just run this command :
+Go to [https://github.com/medcl/logstash-forwarder-java/releases/tag/TCP-Forwarder](https://github.com/medcl/logstash-forwarder-java/releases/tag/TCP-Forwarder)
 
-    java -jar logstash-forwarder-java-X.Y.Z.jar -config /path/to/config/file.json
+1. Download `logstash-forwarder-java.zip`, unzip it
+
+2. Download related JDK to `logstash-forwarder-java` folder, unzip it, `7z x jdk8u232-b09-jre-<OS>.7z`
+
+3. Change `run.sh`, update JDK path
+
+4. Update `config.json`, with your server and file path
 
-For help run :
+5. Run `run.sh`
 
-    java -jar logstash-forwarder-java-X.Y.Z.jar -help
+For more JRE, download from here:
 
-## Differences with logstash-forwarder
+- https://adoptopenjdk.net/releases.html?variant=openjdk8&jvmVariant=hotspot
 
 ### Configuration
 
-The configuration file is the same (json format), but there are a few differences :
+For help run `java -jar logstash-forwarder-java-X.Y.Z.jar -help`:
+
+  - help
   - the ssl ca parameter points to a java [keystore](https://github.com/didfet/logstash-forwarder-java/blob/master/HOWTO-KEYSTORE.md) containing the root certificate of the server, not a PEM file
   - comments are C-style comments
   - multiline support with attributes "pattern", "negate" (true/false) and "what" (previous/next) (version 0.2.5)
   - filtering support with attributes "pattern" and "negate" (true/false) (version 0.2.5)
-
-### Command-line options
-
-Some options are the same :
   - config (but only for a file, not a directory)
   - quiet
   - idle-timeout (renamed idletimeout)
   - spool-size (renamed spoolsize)
   - tail
-  - help
-
-There are a few more options :
   - debug : turn on debug logging level
   - trace : turn on trace logging level
   - signaturelength : size of the block used to compute the checksum

config.json

@@ -0,0 +1,12 @@
+{
+  "network":{
+    "servers":["127.0.0.1:5055"]
+  },
+  "files":[
+    {
+      "paths":["/tmp/123.log"]
+    },{
+      "paths":["/private/tmp/123.log"]
+    }
+  ]
+}

pom.xml

@@ -48,10 +48,10 @@
 							<mainClass>info.fetter.logstashforwarder.Forwarder</mainClass>
 							<addDefaultImplementationEntries>true</addDefaultImplementationEntries>
 						</manifest>
-						<manifestEntries>
-							<Implementation-Build>${prefix.revision}</Implementation-Build>
-							<buildtime>${timestamp}</buildtime>
-						</manifestEntries>
+						<!--<manifestEntries>-->
+							<!--<Implementation-Build>${prefix.revision}</Implementation-Build>-->
+							<!--<buildtime>${timestamp}</buildtime>-->
+						<!--</manifestEntries>-->
 					</archive>
 				</configuration>
 			</plugin>
@@ -95,7 +95,7 @@
 		</plugins>
 		<pluginManagement>
 			<plugins>
-				<!--This plugin's configuration is used to store Eclipse m2e settings 
+				<!--This plugin's configuration is used to store Eclipse m2e settings
 					only. It has no influence on the Maven build itself. -->
 				<plugin>
 					<groupId>org.eclipse.m2e</groupId>
@@ -133,6 +133,11 @@
 		</pluginManagement>
 	</build>
 	<dependencies>
+		<dependency>
+			<groupId>com.alibaba</groupId>
+			<artifactId>fastjson</artifactId>
+			<version>1.2.61</version>
+		</dependency>
 		<dependency>
 			<groupId>commons-io</groupId>
 			<artifactId>commons-io</artifactId>

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

@@ -22,7 +22,7 @@
 import java.util.Map;
 
 public class Event {
-	private Map<String,byte[]> keyValues = new HashMap<String,byte[]>(10);
+	private Map<String,Object> keyValues = new HashMap<String,Object>(10);
 
 	public Event() {
 	}
@@ -39,26 +39,26 @@ public Event(Map<String,String> fields) throws UnsupportedEncodingException {
 		}
 	}
 
-	public Event addField(String key, byte[] value) {
-		keyValues.put(key, value);
-		return this;
-	}
+//	public Event addField(String key, byte[] value) {
+//		keyValues.put(key, value);
+//		return this;
+//	}
 
 	public Event addField(String key, String value) throws UnsupportedEncodingException {
-		keyValues.put(key, value.getBytes());
+		keyValues.put(key, value);
 		return this;
 	}
 
 	public Event addField(String key, long value) throws UnsupportedEncodingException {
-		keyValues.put(key, String.valueOf(value).getBytes());
+		keyValues.put(key, value);
 		return this;
 	}
 
-	public Map<String,byte[]> getKeyValues() {
+	public Map<String,Object> getKeyValues() {
 		return keyValues;
 	}
 
-	public byte[] getValue(String fieldName) {
+	public Object getValue(String fieldName) {
 		return keyValues.get(fieldName);
 	}
 }

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

@@ -46,7 +46,7 @@ public FileReader(int spoolSize) {
 		super(spoolSize);
 	}
 
-	public int readFiles(Collection<FileState> fileList) throws AdapterException {
+	public int readFiles(Collection<FileState> fileList) throws AdapterException, IOException {
 		int eventCount = 0;
 		if(logger.isTraceEnabled()) {
 			logger.trace("Reading " + fileList.size() + " file(s)");
@@ -70,7 +70,7 @@ public int readFiles(Collection<FileState> fileList) throws AdapterException {
 		return eventCount; // Return number of events sent to adapter
 	}
 
-	private int readFile(FileState state, int spaceLeftInSpool) {
+	private int readFile(FileState state, int spaceLeftInSpool) throws IOException {
 		File file = state.getFile();
 		long pointer = state.getPointer();
 		int numberOfEvents = 0;
@@ -120,7 +120,11 @@ private boolean isCompressedFile(FileState state) {
 				}
 			}
 		} catch(IOException e) {
-			logger.warn("Exception raised while reading file : " + state.getFile(), e);
+			try {
+				logger.warn("Exception raised while reading file : " + state.getFile(), e);
+			} catch (IOException e1) {
+				e1.printStackTrace();
+			}
 		}
 		return false;
 	}
@@ -208,7 +212,11 @@ private long readLines(FileState state, int spaceLeftInSpool) {
 			}
 			reader.seek(pos); // Ensure we can re-read if necessary
 		} catch(IOException e) {
-			logger.warn("Exception raised while reading file : " + state.getFile(), e);
+			try {
+				logger.warn("Exception raised while reading file : " + state.getFile(), e);
+			} catch (IOException e1) {
+				e1.printStackTrace();
+			}
 		}
 		return pos;
 	}

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

@@ -83,8 +83,8 @@ private void setFileFromDirectoryAndName() throws FileNotFoundException {
 		}
 	}
 
-	public File getFile() {
-		return file;
+	public File getFile() throws IOException {
+		return file.getCanonicalFile();
 	}
 
 	public long getLastModified() {

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

@@ -224,7 +224,7 @@ private void processModifications() throws IOException {
 	}
 
 	private void addSingleFile(String fileToWatch, Event fields, long deadTime, Multiline multiline, Filter filter) throws Exception {
-		logger.info("Watching file : " + new File(fileToWatch).getCanonicalPath());
+		logger.info("Checking file : " + new File(fileToWatch).getCanonicalPath());
 		String directory = FilenameUtils.getFullPath(fileToWatch);
 		String fileName = FilenameUtils.getName(fileToWatch);
 		IOFileFilter fileFilter = FileFilterUtils.and(
@@ -269,7 +269,7 @@ private void initializeWatchMap(File directory, IOFileFilter fileFilter, Event f
 
 	private void addFileToWatchMap(Map<File,FileState> map, File file, Event fields, Multiline multiline, Filter filter) {
 		try {
-			FileState state = new FileState(file);
+			FileState state = new FileState(file.getCanonicalFile());
 			state.setFields(fields);
 			int signatureLength = (int) (state.getSize() > maxSignatureLength ? maxSignatureLength : state.getSize());
 			state.setSignatureLength(signatureLength);
@@ -278,7 +278,7 @@ private void addFileToWatchMap(Map<File,FileState> map, File file, Event fields,
 			logger.trace("Setting signature of size : " + signatureLength + " on file : " + file + " : " + signature);
 			state.setMultiline(multiline);
 			state.setFilter(filter);
-			map.put(file, state);
+			map.put(file.getCanonicalFile(), state);
 		} catch(IOException e) {
 			logger.error("Caught IOException in addFileToWatchMap : " +
 						 e.getMessage());

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

@@ -17,34 +17,19 @@
  *
  */
 
-import static org.apache.log4j.Level.*;
+import info.fetter.logstashforwarder.config.ConfigurationManager;
+import info.fetter.logstashforwarder.config.FilesSection;
+import info.fetter.logstashforwarder.protocol.TCPClient;
+import info.fetter.logstashforwarder.util.AdapterException;
+import org.apache.commons.cli.*;
+import org.apache.log4j.*;
+import org.apache.log4j.spi.RootLogger;
 
 import java.io.IOException;
 import java.util.List;
 import java.util.Random;
 
-import info.fetter.logstashforwarder.config.ConfigurationManager;
-import info.fetter.logstashforwarder.config.FilesSection;
-import info.fetter.logstashforwarder.protocol.LumberjackClient;
-import info.fetter.logstashforwarder.util.AdapterException;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.GnuParser;
-import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.OptionBuilder;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.apache.log4j.Appender;
-import org.apache.log4j.BasicConfigurator;
-import org.apache.log4j.ConsoleAppender;
-import org.apache.log4j.Layout;
-import org.apache.log4j.Level;
-import org.apache.log4j.Logger;
-import org.apache.log4j.PatternLayout;
-import org.apache.log4j.RollingFileAppender;
-import org.apache.log4j.spi.RootLogger;
+import static org.apache.log4j.Level.*;
 
 public class Forwarder {
 	private static final String SINCEDB = ".logstash-forwarder-java";
@@ -127,7 +112,8 @@ private static void connectToServer() {
 				randomServerIndex = random.nextInt(serverList.size());
 				String[] serverAndPort = serverList.get(randomServerIndex).split(":");
 				logger.info("Trying to connect to " + serverList.get(randomServerIndex));
-				adapter = new LumberjackClient(configManager.getConfig().getNetwork().getSslCA(),serverAndPort[0],Integer.parseInt(serverAndPort[1]), networkTimeout);
+                adapter = new TCPClient(configManager.getConfig().getNetwork().getSslCA(),serverAndPort[0],Integer.parseInt(serverAndPort[1]), networkTimeout);
+//                adapter = new LumberjackClient(configManager.getConfig().getNetwork().getSslCA(),serverAndPort[0],Integer.parseInt(serverAndPort[1]), networkTimeout);
 				fileReader.setAdapter(adapter);
 				inputReader.setAdapter(adapter);
 			} catch(Exception ex) {

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

@@ -23,6 +23,7 @@
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 
@@ -66,15 +67,15 @@ protected void addEvent(String fileName, Event fields, long pos, byte[] line) th
 		Event event = new Event(fields);
 		event.addField("file", fileName)
 		.addField("offset", pos)
-		.addField("line", line)
+		.addField("message", new String(line))
 		.addField("host", hostname);
 		eventList.add(event);
 	}
 
 	protected void addEvent(String fileName, Event fields, long pos, String line) throws IOException {
 		Event event = new Event(fields);
 		event.addField("file", fileName)
-		.addField("offset", pos)
+		.addField("message", pos)
 		.addField("line", line)
 		.addField("host", hostname);
 		eventList.add(event);

src/main/java/info/fetter/logstashforwarder/config/ConfigurationManager.java

@@ -33,7 +33,7 @@ public class ConfigurationManager {
 	public ConfigurationManager(String configFilePath) {
 		this(new File(configFilePath));
 	}
-	
+
 	public ConfigurationManager(File file) {
 		configFile = file;
 		mapper = new ObjectMapper();

src/main/java/info/fetter/logstashforwarder/config/FilesSection.java

@@ -17,6 +17,8 @@
  *
  */
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import org.apache.commons.lang.builder.ToStringBuilder;
@@ -27,8 +29,8 @@
 import java.io.UnsupportedEncodingException;
 
 public class FilesSection {
-	private List<String> paths;
-	private Map<String,String> fields;
+	private List<String> paths=new ArrayList<String>();
+	private Map<String,String> fields=Collections.emptyMap();
 	@JsonProperty("dead time")
 	private String deadTime = "24h";
 	private Multiline multiline;