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

Added debug information.

Browse Source
This commit is contained in:
didfet
2015-03-10 16:20:56 +01:00
parent 25eb1e99f8
commit de012bb84a
1 changed files with 28 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/main/java/info/fetter/logstashforwarder/protocol/LumberjackClient.java
View File

@@ -37,7 +37,11 @@ import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocket; import javax.net.ssl.SSLSocket;
import javax.net.ssl.TrustManagerFactory; import javax.net.ssl.TrustManagerFactory;
import org.apache.commons.io.HexDump;
import org.apache.log4j.Logger;
public class LumberjackClient { public class LumberjackClient {
private final static Logger logger = Logger.getLogger(LumberjackClient.class);
private final static byte PROTOCOL_VERSION = 0x31; private final static byte PROTOCOL_VERSION = 0x31;
private final static byte FRAME_ACK = 0x41; private final static byte FRAME_ACK = 0x41;
private final static byte FRAME_WINDOW_SIZE = 0x57; private final static byte FRAME_WINDOW_SIZE = 0x57;
@@ -73,6 +77,8 @@ public class LumberjackClient {
output = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream())); output = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
input = new DataInputStream(socket.getInputStream()); input = new DataInputStream(socket.getInputStream());
logger.info("Connected to " + server + ":" + port);
} catch(IOException e) { } catch(IOException e) {
throw e; throw e;
} catch(Exception e) { } catch(Exception e) {
@@ -85,6 +91,7 @@ public class LumberjackClient {
output.writeByte(FRAME_WINDOW_SIZE); output.writeByte(FRAME_WINDOW_SIZE);
output.writeInt(size); output.writeInt(size);
output.flush(); output.flush();
logger.debug("Sending window size frame : " + size + " frames");
return 6; return 6;
} }
@@ -121,11 +128,17 @@ public class LumberjackClient {
ByteArrayOutputStream uncompressedBytes = new ByteArrayOutputStream(); ByteArrayOutputStream uncompressedBytes = new ByteArrayOutputStream();
DataOutputStream uncompressedOutput = new DataOutputStream(uncompressedBytes); DataOutputStream uncompressedOutput = new DataOutputStream(uncompressedBytes);
for(Map<String,byte[]> keyValues : keyValuesList) { for(Map<String,byte[]> keyValues : keyValuesList) {
logger.debug("Adding data frame");
sendDataFrame(uncompressedOutput, keyValues); sendDataFrame(uncompressedOutput, keyValues);
} }
uncompressedOutput.close(); uncompressedOutput.close();
Deflater compressor = new Deflater(); Deflater compressor = new Deflater();
compressor.setInput(uncompressedBytes.toByteArray()); byte[] uncompressedData = uncompressedBytes.toByteArray();
logger.debug("Deflating data : " + uncompressedData.length + " bytes");
if(logger.isTraceEnabled()) {
HexDump.dump(uncompressedData, 0, System.out, 0);
}
compressor.setInput(uncompressedData);
compressor.finish(); compressor.finish();
ByteArrayOutputStream compressedBytes = new ByteArrayOutputStream(); ByteArrayOutputStream compressedBytes = new ByteArrayOutputStream();
@@ -135,10 +148,17 @@ public class LumberjackClient {
compressedBytes.write(buffer, 0, count); compressedBytes.write(buffer, 0, count);
} }
compressedBytes.close(); compressedBytes.close();
byte[] compressedData = compressedBytes.toByteArray();
logger.debug("Deflated data : " + compressor.getTotalOut() + " bytes");
if(logger.isTraceEnabled()) {
HexDump.dump(compressedData, 0, System.out, 0);
}
output.writeInt(compressor.getTotalOut()); output.writeInt(compressor.getTotalOut());
output.write(compressedBytes.toByteArray()); output.write(compressedData);
output.flush();
logger.debug("Sending compressed frame : " + keyValuesList.size() + " frames");
return 6 + compressor.getTotalOut(); return 6 + compressor.getTotalOut();
} }
@@ -152,23 +172,26 @@ public class LumberjackClient {
throw new ProtocolException("Frame type should be Ack, received " + frameType); throw new ProtocolException("Frame type should be Ack, received " + frameType);
} }
int sequenceNumber = input.readInt(); int sequenceNumber = input.readInt();
logger.debug("Received ack sequence : " + sequenceNumber);
return sequenceNumber; return sequenceNumber;
} }
public int sendEvents(List<Event> eventList) throws IOException { public int sendEvents(List<Event> eventList) throws IOException {
int beginSequence = sequence; int beginSequence = sequence;
int numberOfEvents = eventList.size(); int numberOfEvents = eventList.size();
logger.info("Sending " + numberOfEvents + " events");
sendWindowSizeFrame(numberOfEvents); sendWindowSizeFrame(numberOfEvents);
List<Map<String,byte[]>> keyValuesList = new ArrayList<Map<String,byte[]>>(numberOfEvents); List<Map<String,byte[]>> keyValuesList = new ArrayList<Map<String,byte[]>>(numberOfEvents);
for(Event event : eventList) { for(Event event : eventList) {
keyValuesList.add(event.getKeyValues()); keyValuesList.add(event.getKeyValues());
} }
sendCompressedFrame(keyValuesList); sendCompressedFrame(keyValuesList);
while(readAckFrame() < sequence) {} while(readAckFrame() < (sequence - 1) ) {}
return sequence - beginSequence; return sequence - beginSequence;
} }
public void close() throws IOException { public void close() throws IOException {
socket.close(); socket.close();
logger.info("Connection to " + server + ":" + port + " closed");
} }
} }