Add Fluency#waitUntilFlushingAllBuffer

komamitsu · Sep 4, 2016 · 99f79ea · 99f79ea
1 parent 9f72905
commit 99f79ea
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 7 deletions.
diff --git a/README.md b/README.md
@@ -134,6 +134,13 @@ fluency.emit(tag, event);
 fluency.close();
 ```
 
+### Wait until all buffer is flushed
+
+```java
+fluency.waitUntilFlushingAllBuffer(MAX_WAIT_BUF_FLUSH);
+fluency.close();
+```
+
 ### Check if Fluency is terminated
 
 ```java

diff --git a/src/main/java/org/komamitsu/fluency/Fluency.java b/src/main/java/org/komamitsu/fluency/Fluency.java
@@ -23,6 +23,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 public class Fluency
         implements Flushable, Closeable
@@ -184,6 +185,21 @@ public boolean isTerminated()
         return flusher.isTerminated();
     }
 
+    public boolean waitUntilFlushingAllBuffer(int maxWaitSeconds)
+            throws InterruptedException
+    {
+        for (int i = 0; i < maxWaitSeconds; i++) {
+            long bufferedDataSize = getBufferedDataSize();
+            LOG.info("Waiting for flushing all buffer: {}", bufferedDataSize);
+            if (getBufferedDataSize() == 0) {
+                return true;
+            }
+            TimeUnit.SECONDS.sleep(1);
+        }
+        LOG.warn("Buffered data still remains: {}", getBufferedDataSize());
+        return false;
+    }
+
     public static class Builder
     {
         private final Sender sender;

diff --git a/src/test/java/org/komamitsu/fluency/FluencyTest.java b/src/test/java/org/komamitsu/fluency/FluencyTest.java
@@ -818,6 +818,7 @@ public void testWithRealFluentd()
             for (Future<Void> future : futures) {
                 future.get(60, TimeUnit.SECONDS);
             }
+            fluency.waitUntilFlushingAllBuffer(60);
         }
         finally {
             fluency.close();
@@ -853,6 +854,7 @@ public void testWithRealMultipleFluentd()
             for (Future<Void> future : futures) {
                 future.get(60, TimeUnit.SECONDS);
             }
+            fluency.waitUntilFlushingAllBuffer(60);
         }
         finally {
             fluency.close();
@@ -881,13 +883,7 @@ public void testWithRealFluentdWithFileBackup()
             for (Future<Void> future : futures) {
                 future.get(60, TimeUnit.SECONDS);
             }
-            for (int i = 0; i < 10; i++) {
-                LOG.debug("BufferedDataSize is {}", fluency.getBufferedDataSize());
-                if (fluency.getBufferedDataSize() == 0) {
-                    break;
-                }
-                TimeUnit.SECONDS.sleep(1);
-            }
+            fluency.waitUntilFlushingAllBuffer(60);
         }
         finally {
             fluency.close();