Febbweiss/logstash-forwarder-java
1
0
Fork 0
mirror of https://github.com/Febbweiss/logstash-forwarder-java.git synced 2026-03-04 22:25:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactored Watcher and Reader.

Browse Source
This commit is contained in:
didfet
2016-08-27 10:45:28 +02:00
parent 118e8efd63
commit adb969518d
2 changed files with 18 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/main/java/info/fetter/logstashforwarder/FileWatcher.java
View File

@@ -45,6 +45,7 @@ public class FileWatcher implements Watcher {
private Map<File,FileState> newWatchMap = new HashMap<File,FileState>(); private Map<File,FileState> newWatchMap = new HashMap<File,FileState>();
private FileState[] savedStates; private FileState[] savedStates;
private Parameters parameters; private Parameters parameters;
private FileReader reader;
public FileWatcher() { public FileWatcher() {
} }
@@ -92,14 +93,6 @@ public class FileWatcher implements Watcher {
printWatchMap(); printWatchMap();
} }
public int readFiles(FileReader reader) throws IOException, AdapterException {
logger.trace("Reading files");
logger.trace("==============");
int numberOfLinesRead = reader.readFiles(oldWatchMap.values());
Registrar.writeStateToJson(parameters.getSincedbFile(),oldWatchMap.values());
return numberOfLinesRead;
}
private void detectModifications() throws IOException { private void detectModifications() throws IOException {
for(File file : newWatchMap.keySet()) { for(File file : newWatchMap.keySet()) {
@@ -356,6 +349,19 @@ public class FileWatcher implements Watcher {
} catch(Exception e) { } catch(Exception e) {
logger.warn("Could not load saved states : " + e.getMessage(), e); logger.warn("Could not load saved states : " + e.getMessage(), e);
} }
reader = new FileReader(parameters.getSpoolSize());
}
public Reader getReader() {
return reader;
}
public int readFiles() throws IOException, AdapterException {
logger.trace("Reading files");
logger.trace("==============");
int numberOfLinesRead = reader.readFiles(oldWatchMap.values());
Registrar.writeStateToJson(parameters.getSincedbFile(),oldWatchMap.values());
return numberOfLinesRead;
} }
} }

12
src/main/java/info/fetter/logstashforwarder/Forwarder.java
View File

@@ -43,9 +43,7 @@ public class Forwarder {
private static Logger logger = Logger.getLogger(Forwarder.class); private static Logger logger = Logger.getLogger(Forwarder.class);
private static ConfigurationManager configManager; private static ConfigurationManager configManager;
private static FileWatcher fileWatcher; private static FileWatcher fileWatcher;
private static FileReader fileReader;
private static InputWatcher inputWatcher; private static InputWatcher inputWatcher;
private static InputReader inputReader;
private static ProtocolAdapter adapter; private static ProtocolAdapter adapter;
private static Random random = new Random(); private static Random random = new Random();
private static Parameters parameters; private static Parameters parameters;
@@ -65,8 +63,6 @@ public class Forwarder {
fileWatcher.addFilesToWatch(files); fileWatcher.addFilesToWatch(files);
} }
fileWatcher.initialize(); fileWatcher.initialize();
fileReader = new FileReader(parameters.getSpoolSize());
inputReader = new InputReader(parameters.getSpoolSize(), System.in);
connectToServer(); connectToServer();
infiniteLoop(); infiniteLoop();
} catch(Exception e) { } catch(Exception e) {
@@ -79,8 +75,8 @@ public class Forwarder {
while(true) { while(true) {
try { try {
fileWatcher.checkFiles(); fileWatcher.checkFiles();
while(fileWatcher.readFiles(fileReader) == parameters.getSpoolSize()); while(fileWatcher.readFiles() == parameters.getSpoolSize());
while(inputWatcher.readStdin(inputReader) == parameters.getSpoolSize()); while(inputWatcher.readFiles() == parameters.getSpoolSize());
Thread.sleep(parameters.getIdleTimeout()); Thread.sleep(parameters.getIdleTimeout());
} catch(AdapterException e) { } catch(AdapterException e) {
logger.error("Lost server connection"); logger.error("Lost server connection");
@@ -109,8 +105,8 @@ public class Forwarder {
String[] serverAndPort = serverList.get(randomServerIndex).split(":"); String[] serverAndPort = serverList.get(randomServerIndex).split(":");
logger.info("Trying to connect to " + serverList.get(randomServerIndex)); 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 LumberjackClient(configManager.getConfig().getNetwork().getSslCA(),serverAndPort[0],Integer.parseInt(serverAndPort[1]), networkTimeout);
fileReader.setAdapter(adapter); fileWatcher.getReader().setAdapter(adapter);
inputReader.setAdapter(adapter); inputWatcher.getReader().setAdapter(adapter);
} catch(Exception ex) { } catch(Exception ex) {
if(logger.isDebugEnabled()) { if(logger.isDebugEnabled()) {
logger.error("Failed to connect to server " + serverList.get(randomServerIndex) + " : ", ex); logger.error("Failed to connect to server " + serverList.get(randomServerIndex) + " : ", ex);