Use Callable instead of Runnable

To get emit returning values, it should use Callable interface.
Verify connected state before #emit.

Author: cosmo0920

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

@@ -2,8 +2,11 @@
 package org.fluentd.logger.sender;
 
 import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 
 import org.fluentd.logger.errorhandler.ErrorHandler;
 import org.fluentd.logger.sender.ExponentialDelayReconnector;
@@ -21,7 +24,7 @@
  */
 public class AsyncRawSocketSender implements Sender {
 
-    private final class EmitRunnable implements Runnable {
+    private final class EmitRunnable implements Callable<Boolean> {
         private final String tag;
         private final Map<String, Object> data;
         private final RawSocketSender sender;
@@ -36,8 +39,11 @@ private EmitRunnable(String tag, Map<String, Object> data,
         }
 
         @Override
-        public void run() {
-            sender.emit(tag, timestamp, data);
+        public Boolean call() {
+            if (!sender.isConnected() && !reconnector.enableReconnection(System.currentTimeMillis()))
+                return false;
+
+            return sender.emit(tag, timestamp, data);
         }
     }
 
@@ -106,9 +112,14 @@ 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;
-        senderTask.execute(new EmitRunnable(tag, data, sender, timestamp));
-
-        return this.isConnected() || reconnector.enableReconnection(System.currentTimeMillis());
+        try {
+            Future<Boolean> result = senderTask.submit(new EmitRunnable(tag, data, sender, timestamp));
+            return result.get();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        } catch (ExecutionException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     @Override