Merge pull request #201 from Leanplum/feat/1355

Author: rossenassenov

AndroidSDKCore/src/main/java/com/leanplum/internal/Request.java

@@ -59,6 +59,7 @@ public class Request {
   private static final long DEVELOPMENT_MIN_DELAY_MS = 100;
   private static final long DEVELOPMENT_MAX_DELAY_MS = 5000;
   private static final long PRODUCTION_DELAY = 60000;
+  private RequestSequenceRecorder requestSequenceRecorder;
   static final int MAX_EVENTS_PER_API_CALL;
   static final String LEANPLUM = "__leanplum__";
   static final String UUID_KEY = "uuid";
@@ -164,6 +165,28 @@ public static void saveToken() {
     SharedPreferencesUtil.commitChanges(editor);
   }
 
+  private static class NoRequestSequenceRecorder implements RequestSequenceRecorder {
+    @Override
+    public void beforeRead() {
+      // No op.
+    }
+
+    @Override
+    public void afterRead() {
+      // No op.
+    }
+
+    @Override
+    public void beforeWrite() {
+      // No op.
+    }
+
+    @Override
+    public void afterWrite() {
+      // No op.
+    }
+  }
+
   public static String appId() {
     return appId;
   }
@@ -177,6 +200,10 @@ public static String userId() {
   }
 
   public Request(String httpMethod, String apiMethod, Map<String, Object> params) {
+    this(httpMethod, apiMethod, params, new NoRequestSequenceRecorder());
+  }
+
+  Request(String httpMethod, String apiMethod, Map<String, Object> params, RequestSequenceRecorder requestSequenceRecorder) {
     this.httpMethod = httpMethod;
     this.apiMethod = apiMethod;
     this.params = params != null ? params : new HashMap<String, Object>();
@@ -187,6 +214,7 @@ public Request(String httpMethod, String apiMethod, Map<String, Object> params)
     // Make sure the Handler is initialized on the main thread.
     OsHandler.getInstance();
     dataBaseIndex = -1;
+    this.requestSequenceRecorder = requestSequenceRecorder;
   }
 
   public static Request get(String apiMethod, Map<String, Object> params) {
@@ -232,6 +260,8 @@ private Map<String, Object> createArgsDictionary() {
 
   private void saveRequestForLater(Map<String, Object> args) {
     try {
+      requestSequenceRecorder.beforeWrite();
+
       synchronized (Request.class) {
         Context context = Leanplum.getContext();
         SharedPreferences preferences = context.getSharedPreferences(
@@ -255,6 +285,8 @@ private void saveRequestForLater(Map<String, Object> args) {
           eventCallbackManager.addCallbacks(this, response, error);
         }
       }
+
+      requestSequenceRecorder.afterWrite();
     } catch (Throwable t) {
       Util.handleException(t);
     }
@@ -549,8 +581,12 @@ private RequestsWithEncoding getRequestsWithEncodedString() {
   }
 
   private void sendRequests() {
+    requestSequenceRecorder.beforeRead();
+
     RequestsWithEncoding requestsWithEncoding = getRequestsWithEncodedString();
 
+    requestSequenceRecorder.afterRead();
+
     List<Map<String, Object>> unsentRequests = requestsWithEncoding.unsentRequests;
     List<Map<String, Object>> requestsToSend = requestsWithEncoding.requestsToSend;
     String jsonEncodedString = requestsWithEncoding.jsonEncodedString;

AndroidSDKCore/src/main/java/com/leanplum/internal/RequestSequenceRecorder.java

@@ -0,0 +1,16 @@
+package com.leanplum.internal;
+
+/** Records request call sequence of read/write operations to database. */
+public interface RequestSequenceRecorder {
+  /** Executes before database read in Request. */
+  void beforeRead();
+
+  /** Executes after database read in Request. */
+  void afterRead();
+
+  /** Executes before database write in Request. */
+  void beforeWrite();
+
+  /** Executes after database write in Request. */
+  void afterWrite();
+}

AndroidSDKTests/src/test/java/com/leanplum/internal/RequestTest.java

@@ -37,10 +37,13 @@
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
 
 import static org.mockito.AdditionalMatchers.not;
 import static org.mockito.Matchers.eq;
@@ -57,6 +60,7 @@
 )
 @PowerMockIgnore({"org.mockito.*", "org.robolectric.*", "org.json.*", "org.powermock.*"})
 public class RequestTest extends TestCase {
+  public final String POST = "POST";
   /**
    * Runs before every test case.
    */
@@ -67,6 +71,42 @@ public void setUp() {
     Leanplum.setApplicationContext(context);
   }
 
+  /** Test that read writes happened sequentially when calling sendNow(). */
+  @Test
+  public void shouldWriteRequestAndSendInSequence() throws InterruptedException {
+    // Given a request.
+    Map<String, Object> params = new HashMap<>();
+    params.put("data1", "value1");
+    params.put("data2", "value2");
+    final ThreadRequestSequenceRecorder threadRequestSequenceRecorder =
+        new ThreadRequestSequenceRecorder();
+    Request request =
+        new Request(POST, Constants.Methods.START, params, threadRequestSequenceRecorder);
+    request.setAppId("fskadfshdbfa", "wee5w4waer422323");
+
+    new Thread(
+            new Runnable() {
+              @Override
+              public void run() {
+                try {
+                  Thread.sleep(100);
+                } catch (InterruptedException e) {
+                  throw new RuntimeException(e);
+                }
+                threadRequestSequenceRecorder.writeSemaphore.release(1);
+              }
+            })
+        .start();
+
+    // When the request is sent.
+    request.sendIfConnected();
+
+    threadRequestSequenceRecorder.testThreadSemaphore.tryAcquire(5000, TimeUnit.MILLISECONDS);
+
+    // Then the request is written to the local db first, and then read and sent.
+    threadRequestSequenceRecorder.assertCallSequence();
+  }
+
   /**
    * Tests the testRemoveIrrelevantBackgroundStartRequests method.
    * <p>
@@ -323,4 +363,54 @@ private List<Map<String, Object>> mockRequests(int requestSize) {
     return requests;
   }
 
+  private static class ThreadRequestSequenceRecorder implements RequestSequenceRecorder {
+    Instant beforeReadTime, afterReadTime, beforeWriteTime, afterWriteTime;
+    final Semaphore writeSemaphore = new Semaphore(0);
+    final Semaphore readSemaphore = new Semaphore(1);
+    final Semaphore testThreadSemaphore = new Semaphore(0);
+
+    @Override
+    public void beforeRead() {
+      try {
+        readSemaphore.tryAcquire(10, TimeUnit.SECONDS);
+        beforeReadTime = Instant.now();
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      } finally {
+        readSemaphore.release();
+      }
+    }
+
+    @Override
+    public void afterRead() {
+      afterReadTime = Instant.now();
+      testThreadSemaphore.release(1);
+    }
+
+    @Override
+    public void beforeWrite() {
+      // since we are blocking on main thread
+      try {
+        writeSemaphore.tryAcquire(10, TimeUnit.SECONDS);
+        Thread.sleep(2000);
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      } finally {
+        writeSemaphore.release();
+      }
+      beforeWriteTime = Instant.now();
+    }
+
+    @Override
+    public void afterWrite() {
+      afterWriteTime = Instant.now();
+    }
+
+    void assertCallSequence() {
+      assertTrue(
+          beforeWriteTime.isBefore(afterWriteTime)
+              && beforeReadTime.isBefore(afterReadTime)
+              && beforeReadTime.isAfter(afterWriteTime));
+    }
+  }
 }