Merge pull request #51 from cookpad/fix_sending_logs

Fix sending logs multiple times in PureeOutput#flushSync()
cookpad · Sep 16, 2015 · 77d076f · 77d076f
2 parents 310ae84 + f215b2d
commit 77d076f
Show file tree

Hide file tree

Showing 4 changed files with 81 additions and 7 deletions.
diff --git a/puree/src/androidTest/java/com/cookpad/puree/outputs/PureeBufferedOutputTest.java b/puree/src/androidTest/java/com/cookpad/puree/outputs/PureeBufferedOutputTest.java
@@ -112,6 +112,35 @@ public void flush() {
         }
     }
 
+    @ParametersAreNonnullByDefault
+    class BufferedOutputAsync extends BufferedOutput {
+
+        int flushCount = 0;
+
+        @Override
+        public void emit(JsonArray jsonArray, final AsyncResult result) {
+            for (final JsonElement item : jsonArray) {
+                new Thread(new Runnable() {
+                    @Override
+                    public void run() {
+                        try {
+                            Thread.sleep(100);
+                        } catch (InterruptedException e) {
+                        }
+                        logs.add(item.toString());
+                        result.success();
+                    }
+                }).start();
+            }
+        }
+
+        @Override
+        public void flush() {
+            flushCount++;
+            super.flush();
+        }
+    }
+
     @ParametersAreNonnullByDefault
     class TruncateBufferedOutput extends BufferedOutput {
 
@@ -294,6 +323,37 @@ public void testPureeBufferedOutput_countEmit() throws Exception {
         Thread.sleep(100);
 
         assertThat(logs.size(), is(lessThanOrEqualTo(3)));
+
+    }
+
+    @Test
+    public void testPureeBufferedOutput_countEmitAsync() throws Exception {
+        BufferedOutputAsync output = new BufferedOutputAsync();
+
+        logger = new PureeConfiguration.Builder(context)
+                .register(PvLog.class, output)
+                .register(FooLog.class, output)
+                .register(BarLog.class, output)
+                .register(BazLog.class, output)
+                .build()
+                .createPureeLogger();
+
+        logger.flush();
+
+        Thread.sleep(100);
+
+        assertThat(logs.size(), is(0));
+
+        logger.send(new FooLog("foo"));
+        logger.send(new BarLog("bar"));
+        logger.send(new BazLog("baz"));
+
+        logger.flush();
+
+        Thread.sleep(1000);
+
+        assertThat(logs.size(), is(3));
+
     }
 
 }
diff --git a/puree/src/main/java/com/cookpad/puree/outputs/PureeBufferedOutput.java b/puree/src/main/java/com/cookpad/puree/outputs/PureeBufferedOutput.java
@@ -10,15 +10,12 @@
 import com.cookpad.puree.storage.Records;
 
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.annotation.ParametersAreNonnullByDefault;
 
 @ParametersAreNonnullByDefault
 public abstract class PureeBufferedOutput extends PureeOutput {
 
-    final AtomicBoolean lock = new AtomicBoolean(false);
-
     RetryableTaskRunner flushTask;
 
     ScheduledExecutorService executor;
@@ -64,12 +61,13 @@ public void run() {
     }
 
     public void flushSync() {
-        if (!lock.compareAndSet(false, true)) {
+        if (!storage.lock()) {
             return;
         }
         final Records records = getRecordsFromStorage();
 
         if (records.isEmpty()) {
+            storage.unlock();
             return;
         }
 
@@ -80,13 +78,13 @@ public void flushSync() {
             public void success() {
                 flushTask.reset();
                 storage.delete(records);
-                lock.set(false);
+                storage.unlock();
             }
 
             @Override
             public void fail() {
                 flushTask.retryLater();
-                lock.set(false);
+                storage.unlock();
             }
         });
     }

diff --git a/puree/src/main/java/com/cookpad/puree/storage/PureeSQLiteStorage.java b/puree/src/main/java/com/cookpad/puree/storage/PureeSQLiteStorage.java
@@ -13,6 +13,8 @@
 import android.text.TextUtils;
 import android.util.Log;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import javax.annotation.ParametersAreNonnullByDefault;
 
 @ParametersAreNonnullByDefault
@@ -32,6 +34,8 @@ public class PureeSQLiteStorage extends SQLiteOpenHelper implements PureeStorage
 
     private final SQLiteDatabase db;
 
+    private final AtomicBoolean lock = new AtomicBoolean(false);
+
     static String databaseName(Context context) {
         // do not share the database file in multi processes
         String processName = ProcessName.getAndroidProcessName(context);
@@ -101,7 +105,7 @@ private int getRecordCount() {
         String query = "SELECT COUNT(*) FROM " + TABLE_NAME;
         Cursor cursor = db.rawQuery(query, null);
         int count = 0;
-        if(cursor.moveToNext()){
+        if (cursor.moveToNext()) {
             count = cursor.getInt(0);
         }
 
@@ -159,4 +163,14 @@ protected void finalize() throws Throwable {
         db.close();
         super.finalize();
     }
+
+    @Override
+    public boolean lock() {
+        return lock.compareAndSet(false, true);
+    }
+
+    @Override
+    public void unlock() {
+        lock.set(false);
+    }
 }
diff --git a/puree/src/main/java/com/cookpad/puree/storage/PureeStorage.java b/puree/src/main/java/com/cookpad/puree/storage/PureeStorage.java
@@ -10,4 +10,6 @@ public interface PureeStorage {
     public void delete(Records records);
     public void truncateBufferedLogs(int maxRecords);
     public void clear();
+    public boolean lock();
+    public void unlock();
 }