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

Author: Mike Kobyakov

src/main/java/org/fluentd/logger/FluentLoggerFactory.java

@@ -19,6 +19,7 @@
 
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Properties;
 import java.util.WeakHashMap;
@@ -33,7 +34,7 @@ public class FluentLoggerFactory {
     private final Map<FluentLogger, String> loggers;
 
     public FluentLoggerFactory() {
-        loggers = new WeakHashMap<FluentLogger, String>();
+        loggers = Collections.synchronizedMap(new WeakHashMap<FluentLogger, String>());
     }
 
     public FluentLogger getLogger(String tagPrefix) {
@@ -48,7 +49,7 @@ public FluentLogger getLogger(String tagPrefix, String host, int port, int timeo
         return getLogger(tagPrefix, host, port, timeout, bufferCapacity, new ExponentialDelayReconnector());
     }
 
-    public synchronized FluentLogger getLogger(String tagPrefix, String host, int port, int timeout, int bufferCapacity,
+    public FluentLogger getLogger(String tagPrefix, String host, int port, int timeout, int bufferCapacity,
             Reconnector reconnector) {
         String key = String.format("%s_%s_%d_%d_%d", new Object[] { tagPrefix, host, port, timeout, bufferCapacity });
 
@@ -80,6 +81,30 @@ public synchronized FluentLogger getLogger(String tagPrefix, String host, int po
         return logger;
     }
 
+    public FluentLogger getLogger(String tagPrefix, String host, int port, int timeout, int bufferCapacity,
+            Sender sender) {
+        if (sender == null) {
+            return getLogger(tagPrefix, host, port, timeout, bufferCapacity);
+        }
+        String key = String.format("%s_%s_%d_%d_%d_%s", new Object[] { tagPrefix, host, port, timeout, bufferCapacity, sender == null ? "null" : sender .getName() });
+        if (loggers.containsKey(key)) {
+            for (Map.Entry<FluentLogger, String> entry : loggers.entrySet()) {
+                if (entry.getValue().equals(key)) {
+                    FluentLogger found = entry.getKey();
+                    if(found != null) {
+                        return found;
+                    }
+                    break;
+                }
+            }
+            return getLogger(tagPrefix, host, port, timeout, bufferCapacity);
+        } else {
+            FluentLogger logger = new FluentLogger(tagPrefix, sender);
+            loggers.put(logger, key);
+            return logger;
+        }
+    }
+
     @SuppressWarnings("unchecked")
     private Sender createSenderInstance(final String className, final Object[] params) throws ClassNotFoundException,
             SecurityException, NoSuchMethodException, IllegalArgumentException, InstantiationException,

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

@@ -0,0 +1,113 @@
+
+package org.fluentd.logger.sender;
+
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.fluentd.logger.errorhandler.ErrorHandler;
+import org.fluentd.logger.sender.ExponentialDelayReconnector;
+import org.fluentd.logger.sender.RawSocketSender;
+import org.fluentd.logger.sender.Reconnector;
+import org.fluentd.logger.sender.Sender;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * 
+ * @author mkobyakov
+ *
+ */
+public class AsyncRawSocketSender implements Sender {
+
+    private RawSocketSender sender;
+    private Reconnector reconnector;
+
+    @SuppressWarnings("unused")
+    private static final Logger logger = LoggerFactory.getLogger(AsyncRawSocketSender.class);
+
+    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);
+    }
+
+    public AsyncRawSocketSender(String host, int port) {
+        this(host, port, 3 * 1000, 8 * 1024 * 1024);
+    }
+
+    public AsyncRawSocketSender(String host, int port, int timeout,
+            int bufferCapacity) {
+        this(host, port, timeout, bufferCapacity,
+                new ExponentialDelayReconnector());
+    }
+
+    public AsyncRawSocketSender(String host, int port, int timeout,
+            int bufferCapacity, Reconnector reconnector) {
+        this.reconnector = reconnector;
+        this.sender = new RawSocketSender(host, port, timeout, bufferCapacity,
+                                          reconnector);
+    }
+
+    @Override
+    public synchronized void flush() {
+        final RawSocketSender sender = this.sender;
+        flusher.execute(new Runnable() {
+            @Override
+            public void run() {
+                sender.flush();
+            }
+        });
+    }
+
+    @Override
+    public void close() {
+        sender.close();
+    }
+
+    @Override
+    public boolean emit(String tag, Map<String, Object> data) {
+        return emit(tag, System.currentTimeMillis() / 1000, 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);
+            }
+        });
+
+        return sender.isConnected() || reconnector.enableReconnection(System.currentTimeMillis());
+    }
+
+    @Override
+    public String getName() {
+        return sender.getName();
+    }
+
+    @Override
+    public boolean isConnected() {
+        return sender.isConnected();
+    }
+
+    @Override
+    public void setErrorHandler(ErrorHandler errorHandler) {
+        if (errorHandler == null) {
+            throw new IllegalArgumentException("errorHandler is null");
+        }
+
+        this.errorHandler = errorHandler;
+    }
+
+    @Override
+    public void removeErrorHandler() {
+        this.errorHandler = DEFAULT_ERROR_HANLDER;
+    }
+}