Remove Sender#getBuffer and make RawSocketSender#getBuffer thread-safe

Author: komamitsu

src/main/java/org/fluentd/logger/sender/NullSender.java

@@ -38,11 +38,6 @@ public boolean emit(String tag, long timestamp, Map<String, Object> data) {
     public void flush() {
     }
 
-    @Override
-    public byte[] getBuffer() {
-        return new byte[0];
-    }
-
     @Override
     public void close() {
     }

src/main/java/org/fluentd/logger/sender/RawSocketSender.java

@@ -91,6 +91,7 @@ private void reconnect(boolean forceReconnection) throws IOException {
         }
     }
 
+    @Override
     public void close() {
         // close output stream
         if (out != null) {
@@ -113,10 +114,12 @@ public void close() {
         }
     }
 
+    @Override
     public boolean emit(String tag, Map<String, Object> data) {
         return emit(tag, System.currentTimeMillis() / 1000, data);
     }
 
+    @Override
     public boolean emit(String tag, long timestamp, Map<String, Object> data) {
         return emit(new Event(tag, timestamp, data));
     }
@@ -171,6 +174,7 @@ private synchronized boolean send(byte[] bytes) {
         return true;
     }
 
+    @Override
     public synchronized void flush() {
         try {
             // check whether connection is established or not
@@ -186,7 +190,7 @@ public synchronized void flush() {
         }
     }
 
-    public byte[] getBuffer() {
+    synchronized byte[] getBuffer() {
         int len = pendings.position();
         pendings.position(0);
         byte[] ret = new byte[len];
@@ -198,6 +202,7 @@ private void clearBuffer() {
         pendings.clear();
     }
 
+    @Override
     public String getName() {
         return name;
     }

src/main/java/org/fluentd/logger/sender/Sender.java

@@ -26,9 +26,7 @@ public interface Sender {
 
     void flush();
 
-    public byte[] getBuffer();
+    void close();
 
-    public void close();
-
-    public String getName();
+    String getName();
 }

src/test/java/org/fluentd/logger/sender/TestSenderFluentdDownOperation.java

@@ -30,7 +30,7 @@ public void testFluentdDownOperation01() throws Exception {
         long timestamp = System.currentTimeMillis() / 1000;
 
         // start senders
-        Sender sender = new RawSocketSender("localhost", port);
+        RawSocketSender sender = new RawSocketSender("localhost", port);
         Map<String, Object> data = new HashMap<String, Object>();
         data.put("t1k1", "t1v1");
         data.put("t1k2", "t1v2");
@@ -75,7 +75,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
         server.start();
 
         // start senders
-        Sender sender = new RawSocketSender("localhost", port);
+        RawSocketSender sender = new RawSocketSender("localhost", port);
 
         // server close
         server.close();