feat: Add submitEvent API with callbacks (#2934)

Author: sdhuka

aws-android-sdk-pinpoint/src/main/java/com/amazonaws/mobileconnectors/pinpoint/analytics/AnalyticsClient.java

@@ -15,12 +15,15 @@
 
 package com.amazonaws.mobileconnectors.pinpoint.analytics;
 
+import androidx.core.util.Consumer;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import com.amazonaws.logging.Log;
 import com.amazonaws.logging.LogFactory;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import org.json.JSONArray;
 import org.json.JSONException;
 import org.json.JSONObject;
@@ -169,6 +172,28 @@ public void submitEvents() {
         eventRecorder.submitEvents();
     }
 
+    /**
+     * Submit all recorded events and returns all the successfully submitted events.
+     * If the device is off line, this is a no-op. See
+     * {@link PinpointConfiguration}
+     * for customizing which Internet connection the SDK can submit on.
+     *
+     * @param onSuccess Callback to return successfully submitted events.
+     * @param onError Callback to return error.
+     */
+    public void submitEvents(
+        Consumer<List<AnalyticsEvent>> onSuccess,
+        Consumer<Exception> onError
+    ) {
+        log.info("Submitting events.");
+        try {
+            Future<List<AnalyticsEvent>> result = eventRecorder.submitEventsWithResult();
+            onSuccess.accept(result.get());
+        } catch (InterruptedException | ExecutionException exception) {
+            onError.accept(exception);
+        }
+    }
+
     /**
      * Adds the specified attribute to all subsequently created events Note: The
      * maximum allowed attributes and metrics on a single event is 40. Attempts

aws-android-sdk-pinpoint/src/main/java/com/amazonaws/mobileconnectors/pinpoint/internal/event/EventRecorder.java

@@ -26,7 +26,9 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -250,6 +252,15 @@ public void run() {
         });
     }
 
+    public Future<List<AnalyticsEvent>> submitEventsWithResult() {
+        return submissionRunnableQueue.submit(new Callable<List<AnalyticsEvent>>() {
+            @Override
+            public List<AnalyticsEvent> call() throws Exception {
+                return processEvents();
+            }
+        });
+    }
+
     /**
      * Reads events of maximum of KEY_MAX_SUBMISSION_SIZE size.
      * The default max request size is DEFAULT_MAX_SUBMISSION_SIZE.
@@ -302,18 +313,18 @@ public List<JSONObject> getAllEvents() {
         return events;
     }
 
-    void processEvents() {
+    List<AnalyticsEvent> processEvents() {
         final long start = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());
 
         Cursor cursor = null;
-
+        List<AnalyticsEvent> result = new ArrayList<>();
         try {
             cursor = dbUtil.queryAllEvents();
 
             if (!cursor.moveToFirst()) {
                 // if the cursor is empty there is nothing to do.
                 log.info("No events available to submit.");
-                return;
+                return result;
             }
 
             int submissions = 0;
@@ -336,6 +347,8 @@ void processEvents() {
                     submissions++;
                 }
 
+                //Add all successfully submitted events to result
+                result.addAll(getSuccessfullySyncedEvents(events, batchIdsAndSizeToDelete));
                 // Delete events from the local database. At this point batchIdsAndSizeToDelete
                 // reflects the set of events that can be deleted from the local database.
                 for (Integer id : batchIdsAndSizeToDelete.keySet()) {
@@ -351,12 +364,28 @@ void processEvents() {
             } while (cursor.moveToNext());
 
             log.info(String.format(Locale.US, "Time of attemptDelivery: %d",
-                    TimeUnit.NANOSECONDS.toMillis(System.nanoTime()) - start));
+                TimeUnit.NANOSECONDS.toMillis(System.nanoTime()) - start));
+        } catch (JSONException jsonException) {
+            log.error("Failed to parse to event object", jsonException);
         } finally {
             if (cursor != null) {
                 cursor.close();
             }
         }
+        log.info(String.format("Submitted %s events", result.size()));
+        return result;
+    }
+
+    private List<AnalyticsEvent> getSuccessfullySyncedEvents(JSONArray events,
+                                                             HashMap<Integer, Integer> batchIdsAndSizeToDelete)
+        throws JSONException {
+        List<AnalyticsEvent> result = new ArrayList<>();
+        for (int i = 0; i<events.length(); i++) {
+            if (batchIdsAndSizeToDelete.containsKey(events.getJSONObject(i).getInt(DATABASE_ID_KEY))) {
+                result.add(AnalyticsEvent.translateToEvent(events.getJSONObject(i)));
+            }
+        }
+        return result;
     }
 
     private void submitEventsAndEndpoint(final JSONArray eventArray,