add getLogger(..., Sender) method to the factory; add an asynchronous wrapper around RawSocketSender

Author: Mike Kobyakov

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

@@ -14,12 +14,46 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * 
- * @author mkobyakov
+ * An asynchronous wrapper around RawSocketSender
+ *
+ * @author mxk
  *
  */
 public class AsyncRawSocketSender implements Sender {
 
+    private final class EmitRunnable implements Runnable {
+        private final String tag;
+        private final Map<String, Object> data;
+        private final RawSocketSender sender;
+        private final long timestamp;
+
+        private EmitRunnable(String tag, Map<String, Object> data,
+                             RawSocketSender sender, long timestamp) {
+            this.tag = tag;
+            this.data = data;
+            this.sender = sender;
+            this.timestamp = timestamp;
+        }
+
+        @Override
+        public void run() {
+            sender.emit(tag, timestamp, data);
+        }
+    }
+
+    private final class FlushRunnable implements Runnable {
+        private final RawSocketSender sender;
+
+        private FlushRunnable(RawSocketSender sender) {
+            this.sender = sender;
+        }
+
+        @Override
+        public void run() {
+            sender.flush();
+        }
+    }
+
     private RawSocketSender sender;
     private Reconnector reconnector;
 
@@ -28,10 +62,6 @@ public class AsyncRawSocketSender implements Sender {
 
     private static final ExecutorService flusher = Executors.newSingleThreadExecutor();
 
-    private static final ErrorHandler DEFAULT_ERROR_HANLDER = new ErrorHandler() {};
-
-    private ErrorHandler errorHandler = DEFAULT_ERROR_HANLDER;
-
     public AsyncRawSocketSender() {
         this("localhost", 24224);
     }
@@ -41,27 +71,22 @@ public AsyncRawSocketSender(String host, int port) {
     }
 
     public AsyncRawSocketSender(String host, int port, int timeout,
-            int bufferCapacity) {
+                                int bufferCapacity) {
         this(host, port, timeout, bufferCapacity,
                 new ExponentialDelayReconnector());
     }
 
     public AsyncRawSocketSender(String host, int port, int timeout,
-            int bufferCapacity, Reconnector reconnector) {
+                                int bufferCapacity, Reconnector reconnector) {
         this.reconnector = reconnector;
         this.sender = new RawSocketSender(host, port, timeout, bufferCapacity,
-                                          reconnector);
+                reconnector);
     }
 
     @Override
     public synchronized void flush() {
         final RawSocketSender sender = this.sender;
-        flusher.execute(new Runnable() {
-            @Override
-            public void run() {
-                sender.flush();
-            }
-        });
+        flusher.execute(new FlushRunnable(sender));
     }
 
     @Override
@@ -77,12 +102,7 @@ public boolean emit(String tag, Map<String, Object> data) {
     @Override
     public boolean emit(final String tag, final long timestamp, final Map<String, Object> data) {
         final RawSocketSender sender = this.sender;
-        flusher.execute(new Runnable() {
-            @Override
-            public void run() {
-                sender.emit(tag, timestamp, data);
-            }
-        });
+        flusher.execute(new EmitRunnable(tag, data, sender, timestamp));
 
         return sender.isConnected() || reconnector.enableReconnection(System.currentTimeMillis());
     }
@@ -99,15 +119,11 @@ public boolean isConnected() {
 
     @Override
     public void setErrorHandler(ErrorHandler errorHandler) {
-        if (errorHandler == null) {
-            throw new IllegalArgumentException("errorHandler is null");
-        }
-
-        this.errorHandler = errorHandler;
+        sender.setErrorHandler(errorHandler);
     }
 
     @Override
     public void removeErrorHandler() {
-        this.errorHandler = DEFAULT_ERROR_HANLDER;
+        sender.removeErrorHandler();
     }
-}
+}