Added semaphores

Author: Sayaan Saha

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

@@ -45,7 +45,6 @@
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.HashMap;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Stack;
@@ -60,7 +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 Waiter waiter;
+  private RequestSequence requestSequence;
   static final int MAX_EVENTS_PER_API_CALL;
   static final String LEANPLUM = "__leanplum__";
   static final String UUID_KEY = "uuid";
@@ -166,7 +165,7 @@ public static void saveToken() {
     SharedPreferencesUtil.commitChanges(editor);
   }
 
-  private static class NoWaiter implements Waiter {
+  private static class NoRequestSequence implements RequestSequence {
     @Override
     public void beforeRead() {
       // No op.
@@ -201,10 +200,10 @@ public static String userId() {
   }
 
   public Request(String httpMethod, String apiMethod, Map<String, Object> params) {
-    this(httpMethod, apiMethod, params, new NoWaiter());
+    this(httpMethod, apiMethod, params, new NoRequestSequence());
   }
 
-  Request(String httpMethod, String apiMethod, Map<String, Object> params, Waiter waiter) {
+  Request(String httpMethod, String apiMethod, Map<String, Object> params, RequestSequence requestSequence) {
     this.httpMethod = httpMethod;
     this.apiMethod = apiMethod;
     this.params = params != null ? params : new HashMap<String, Object>();
@@ -215,7 +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.waiter = waiter;
+    this.requestSequence = requestSequence;
   }
 
   public static Request get(String apiMethod, Map<String, Object> params) {
@@ -261,7 +260,8 @@ private Map<String, Object> createArgsDictionary() {
 
   private void saveRequestForLater(Map<String, Object> args) {
     try {
-      waiter.beforeWrite();
+      requestSequence.beforeWrite();
+
       synchronized (Request.class) {
         Context context = Leanplum.getContext();
         SharedPreferences preferences = context.getSharedPreferences(
@@ -284,7 +284,7 @@ private void saveRequestForLater(Map<String, Object> args) {
         if (response != null || error != null && !Util.isConnected()) {
           eventCallbackManager.addCallbacks(this, response, error);
         }
-        waiter.afterWrite();
+        requestSequence.afterWrite();
       }
     } catch (Throwable t) {
       Util.handleException(t);
@@ -580,7 +580,7 @@ private RequestsWithEncoding getRequestsWithEncodedString() {
   }
 
   private void sendRequests() {
-    waiter.beforeRead();
+    requestSequence.beforeRead();
     RequestsWithEncoding requestsWithEncoding = getRequestsWithEncodedString();
 
     List<Map<String, Object>> unsentRequests = requestsWithEncoding.unsentRequests;
@@ -647,7 +647,7 @@ private void sendRequests() {
             parseResponseBody(responseBody, requestsToSend, errorException, unsentRequests.size());
           }
         }
-        waiter.afterRead();
+        requestSequence.afterRead();
       } catch (JSONException e) {
         Log.e("Error parsing JSON response: " + e.toString() + "\n" + Log.getStackTraceString(e));
         deleteSentRequests(unsentRequests.size());

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

@@ -0,0 +1,18 @@
+package com.leanplum.internal;
+
+/**
+ * Public interface for during the call sequence of database
+ * read/write operations.
+ *
+ * @author sayaan
+ */
+
+public interface RequestSequence {
+  void beforeRead();
+
+  void afterRead();
+
+  void beforeWrite();
+
+  void afterWrite();
+}

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

@@ -42,6 +42,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Semaphore;
 
 import static org.mockito.AdditionalMatchers.not;
 import static org.mockito.Matchers.eq;
@@ -58,6 +59,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.
    */
@@ -72,14 +74,22 @@ public void setUp() {
    * Test that read writes happened sequentially when calling sendNow().
    */
   @Test
-  public void testSequenceOfDatabaseOperations() {
+  public void shouldWriteRequestAndSendInSequence() {
+    // Given a request.
     Map<String, Object> params = new HashMap<>();
     params.put("data1", "value1");
     params.put("data2", "value2");
-    ThreadWaiter waiter = new ThreadWaiter();
-    Request request = new Request("POST", Constants.Methods.START, params, waiter);
+    ThreadRequestSequence waiter = new ThreadRequestSequence();
+    Request request = new Request(POST, Constants.Methods.START, params, waiter);
     request.setAppId("fskadfshdbfa", "wee5w4waer422323");
+
+    //block the write
+    waiter.setBlockWrite(true);
+
+    // When the request is sent.
     request.sendIfConnected();
+
+    // When the request is sent.
     waiter.assertCallSequence();
   }
   /**
@@ -338,30 +348,53 @@ private List<Map<String, Object>> mockRequests(int requestSize) {
     return requests;
   }
 
-  private static class ThreadWaiter implements Waiter {
-    Instant t1, t2, t3, t4;
+  private static class ThreadRequestSequence implements RequestSequence {
+    Instant beforeReadTime, afterReadTime, beforeWriteTime, afterWriteTime;
+    final Semaphore semaphore = new Semaphore(1);
+    private boolean blockWrite = false;
+
     @Override
     public void beforeRead() {
-      t1 = Instant.now();
+      try {
+        semaphore.tryAcquire();
+        beforeReadTime = Instant.now();
+      } finally {
+        semaphore.release();
+      }
     }
 
     @Override
     public void afterRead() {
-      t2 = Instant.now();
+      afterReadTime = Instant.now();
     }
 
     @Override
     public void beforeWrite() {
-      t3 = Instant.now();
+      // since we are blocking on main thread
+      if (blockWrite) {
+        try {
+          semaphore.tryAcquire();
+          Thread.sleep(2000);
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+        } finally {
+          semaphore.release();
+        }
+      }
+      beforeWriteTime = Instant.now();
     }
 
     @Override
     public void afterWrite() {
-      t4 = Instant.now();
+      afterWriteTime = Instant.now();
+    }
+
+    public void setBlockWrite(boolean value) {
+      this.blockWrite = value;
     }
 
     public void assertCallSequence() {
-      assertTrue(t4.isBefore(t1));
+      assertTrue(afterWriteTime.isBefore(beforeReadTime));
     }
   }
 }