Submitter module

api/.gitignore

@@ -1 +1,3 @@
 /build
+.classpath
+.settings

backoff/.gitignore

@@ -2,3 +2,5 @@
 .gradle
 *.iml
 .DS_Store
+.classpath
+.settings

build.gradle

@@ -145,7 +145,7 @@ dependencies {
         return candidates.find { findProject(it) != null }
     }
 
-    ['main', 'logger', 'events', 'events-domain', 'api', 'http-api', 'http', 'fallback', 'backoff', 'tracker'].each { moduleName ->
+    ['main', 'logger', 'events', 'events-domain', 'api', 'http-api', 'http', 'fallback', 'backoff', 'tracker', 'submitter'].each { moduleName ->
         def resolvedPath = resolveProjectPath(moduleName)
         if (resolvedPath != null) {
             include project(resolvedPath)

events-domain/.gitignore

@@ -1 +1,3 @@
 /build
+.classpath
+.settings

events/.gitignore

@@ -1 +1,3 @@
-/build
+/build
+.classpath
+.settings

fallback/.gitignore

@@ -2,3 +2,5 @@
 .gradle
 *.iml
 .DS_Store
+.classpath
+.settings

gradle/common-android-library.gradle

@@ -14,6 +14,10 @@ tasks.withType(JavaCompile).configureEach {
     options.compilerArgs.add('-parameters')
 }
 
+tasks.withType(Test).configureEach {
+    maxParallelForks = Runtime.runtime.availableProcessors().intdiv(2) ?: 1
+}
+
 def kotlinCompileClass = null
 try {
     kotlinCompileClass = Class.forName('org.jetbrains.kotlin.gradle.tasks.KotlinCompile')

http-api/.gitignore

@@ -1 +1,3 @@
 /build
+.classpath
+.settings

http/.gitignore

@@ -1 +1,3 @@
 /build
+.classpath
+.settings

logger/.gitignore

@@ -2,3 +2,5 @@
 .gradle
 *.iml
 .DS_Store
+.classpath
+.settings

main/.gitignore

@@ -2,3 +2,5 @@
 .gradle
 *.iml
 .DS_Store
+.classpath
+.settings

main/build.gradle

@@ -56,6 +56,8 @@ dependencies {
     api clientModuleProject('fallback')
     implementation clientModuleProject('backoff')
     implementation clientModuleProject('tracker')
+    api clientModuleProject('submitter')
+
     // Internal module dependencies
     implementation clientModuleProject('http')
     implementation clientModuleProject('events-domain')

main/src/main/java/io/split/android/client/dtos/Event.java

@@ -3,7 +3,7 @@
 import com.google.gson.annotations.JsonAdapter;
 import com.google.gson.annotations.SerializedName;
 
-import io.split.android.client.storage.common.InBytesSizable;
+import io.split.android.client.submitter.InBytesSizable;
 import io.split.android.client.utils.deserializer.EventDeserializer;
 
 @JsonAdapter(EventDeserializer.class)

main/src/main/java/io/split/android/client/dtos/KeyImpression.java

@@ -6,7 +6,7 @@
 import java.util.Objects;
 
 import io.split.android.client.service.ServiceConstants;
-import io.split.android.client.storage.common.InBytesSizable;
+import io.split.android.client.submitter.InBytesSizable;
 import io.split.android.client.impressions.Impression;
 
 public class KeyImpression implements InBytesSizable, Identifiable {

main/src/main/java/io/split/android/client/service/HttpRecorderSubmitterAdapter.java

@@ -0,0 +1,28 @@
+package io.split.android.client.service;
+
+import androidx.annotation.NonNull;
+
+import io.split.android.client.service.http.HttpRecorder;
+import io.split.android.client.service.http.HttpRecorderException;
+import io.split.android.client.service.http.HttpStatus;
+import io.split.android.client.submitter.RecorderException;
+import io.split.android.client.submitter.RecorderSubmitter;
+
+public class HttpRecorderSubmitterAdapter<T> implements RecorderSubmitter<T> {
+    private final HttpRecorder<T> mHttpRecorder;
+
+    public HttpRecorderSubmitterAdapter(@NonNull HttpRecorder<T> httpRecorder) {
+        mHttpRecorder = httpRecorder;
+    }
+
+    @Override
+    public void execute(@NonNull T data) throws RecorderException {
+        try {
+            mHttpRecorder.execute(data);
+        } catch (HttpRecorderException e) {
+            Integer httpStatus = e.getHttpStatus();
+            boolean retryable = !HttpStatus.isNotRetryable(HttpStatus.fromCode(httpStatus));
+            throw new RecorderException(e.getMessage(), httpStatus, retryable);
+        }
+    }
+}

main/src/main/java/io/split/android/client/service/TelemetryRecorderAdapter.java

@@ -0,0 +1,33 @@
+package io.split.android.client.service;
+
+import androidx.annotation.NonNull;
+
+import io.split.android.client.submitter.RecorderTelemetry;
+import io.split.android.client.telemetry.model.OperationType;
+import io.split.android.client.telemetry.storage.TelemetryRuntimeProducer;
+
+public class TelemetryRecorderAdapter implements RecorderTelemetry {
+    private final TelemetryRuntimeProducer mTelemetryProducer;
+    private final OperationType mOperationType;
+
+    public TelemetryRecorderAdapter(@NonNull TelemetryRuntimeProducer telemetryProducer,
+                                    @NonNull OperationType operationType) {
+        mTelemetryProducer = telemetryProducer;
+        mOperationType = operationType;
+    }
+
+    @Override
+    public void recordSuccess(long timestamp) {
+        mTelemetryProducer.recordSuccessfulSync(mOperationType, timestamp);
+    }
+
+    @Override
+    public void recordError(Integer httpStatus) {
+        mTelemetryProducer.recordSyncError(mOperationType, httpStatus);
+    }
+
+    @Override
+    public void recordLatency(long latencyMs) {
+        mTelemetryProducer.recordSyncLatency(mOperationType, latencyMs);
+    }
+}

main/src/main/java/io/split/android/client/service/events/EventsRecorderTask.java

@@ -1,116 +1,42 @@
 package io.split.android.client.service.events;
 
-import static io.split.android.client.utils.Utils.checkNotNull;
-import static io.split.android.client.utils.Utils.partition;
-
 import androidx.annotation.NonNull;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
 import io.split.android.client.dtos.Event;
-import io.split.android.client.service.executor.SplitTask;
-import io.split.android.client.service.executor.SplitTaskExecutionInfo;
-import io.split.android.client.service.executor.SplitTaskExecutionStatus;
+import io.split.android.client.service.HttpRecorderSubmitterAdapter;
+import io.split.android.client.service.TelemetryRecorderAdapter;
 import io.split.android.client.service.executor.SplitTaskType;
 import io.split.android.client.service.http.HttpRecorder;
-import io.split.android.client.service.http.HttpRecorderException;
-import io.split.android.client.service.http.HttpStatus;
 import io.split.android.client.storage.events.PersistentEventsStorage;
+import io.split.android.client.submitter.RecorderTask;
 import io.split.android.client.telemetry.model.OperationType;
 import io.split.android.client.telemetry.storage.TelemetryRuntimeProducer;
-import io.split.android.client.utils.logger.Logger;
 
-public class EventsRecorderTask implements SplitTask {
-    public final static int FAILING_CHUNK_SIZE = 20;
-    private final PersistentEventsStorage mPersistentEventsStorage;
-    private final HttpRecorder<List<Event>> mHttpRecorder;
-    private final EventsRecorderTaskConfig mConfig;
-    private final TelemetryRuntimeProducer mTelemetryRuntimeProducer;
+public class EventsRecorderTask extends RecorderTask<Event, List<Event>> {
+
+    public static final int FAILING_CHUNK_SIZE = 20;
 
     public EventsRecorderTask(@NonNull HttpRecorder<List<Event>> httpRecorder,
-                              @NonNull PersistentEventsStorage persistentEventsStorage,
+                              @NonNull PersistentEventsStorage storage,
                               @NonNull EventsRecorderTaskConfig config,
                               @NonNull TelemetryRuntimeProducer telemetryRuntimeProducer) {
-        mHttpRecorder = checkNotNull(httpRecorder);
-        mPersistentEventsStorage = checkNotNull(persistentEventsStorage);
-        mConfig = checkNotNull(config);
-        mTelemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
+        super(storage,
+                new HttpRecorderSubmitterAdapter<>(httpRecorder),
+                config.getEventsPerPush(),
+                SplitTaskType.EVENTS_RECORDER,
+                new TelemetryRecorderAdapter(telemetryRuntimeProducer, OperationType.EVENTS),
+                FAILING_CHUNK_SIZE);
     }
 
     @Override
-    @NonNull
-    public SplitTaskExecutionInfo execute() {
-        SplitTaskExecutionStatus status = SplitTaskExecutionStatus.SUCCESS;
-        int nonSentRecords = 0;
-        long nonSentBytes = 0;
-        List<Event> events;
-        List<Event> failingEvents = new ArrayList<>();
-        boolean doNotRetry = false;
-        do {
-            events = mPersistentEventsStorage.pop(mConfig.getEventsPerPush());
-            if (events.size() > 0) {
-                long startTime = System.currentTimeMillis();
-                long latency = 0;
-                try {
-                    Logger.d("Posting %d Split events", events.size());
-                    mHttpRecorder.execute(events);
-
-                    long now = System.currentTimeMillis();
-                    latency = now - startTime;
-                    mTelemetryRuntimeProducer.recordSuccessfulSync(OperationType.EVENTS, now);
-
-                    mPersistentEventsStorage.delete(events);
-                    Logger.d("%d split events sent", events.size());
-                } catch (HttpRecorderException e) {
-                    status = SplitTaskExecutionStatus.ERROR;
-                    nonSentRecords += mConfig.getEventsPerPush();
-                    nonSentBytes += sumEventBytes(events);
-                    Logger.e("Event recorder task: Some events couldn't be sent" +
-                            "Saving to send them in a new iteration: " +
-                            e.getLocalizedMessage());
-                    failingEvents.addAll(events);
-
-                    mTelemetryRuntimeProducer.recordSyncError(OperationType.EVENTS, e.getHttpStatus());
-
-                    if (HttpStatus.isNotRetryable(e.getHttpStatus())) {
-                        doNotRetry = true;
-                        break;
-                    }
-                } finally {
-                    mTelemetryRuntimeProducer.recordSyncLatency(OperationType.EVENTS, latency);
-                }
-            }
-        } while (events.size() == mConfig.getEventsPerPush());
-
-        // Update events by chunks to avoid sqlite errors
-        List<List<Event>> failingChunks = partition(failingEvents, FAILING_CHUNK_SIZE);
-        for (List<Event> chunk : failingChunks) {
-            mPersistentEventsStorage.setActive(chunk);
-        }
-
-        if (status == SplitTaskExecutionStatus.ERROR) {
-            Map<String, Object> data = new HashMap<>();
-            data.put(SplitTaskExecutionInfo.NON_SENT_RECORDS, nonSentRecords);
-            data.put(SplitTaskExecutionInfo.NON_SENT_BYTES, nonSentBytes);
-            if (doNotRetry) {
-                data.put(SplitTaskExecutionInfo.DO_NOT_RETRY, true);
-            }
-
-            return SplitTaskExecutionInfo.error(
-                    SplitTaskType.EVENTS_RECORDER, data);
-        }
-        return SplitTaskExecutionInfo.success(SplitTaskType.EVENTS_RECORDER);
+    protected List<Event> transformForSubmission(List<Event> items) {
+        return items;
     }
 
-    private long sumEventBytes(List<Event> events) {
-        long totalBytes = 0;
-        for (Event event : events) {
-            totalBytes += event.getSizeInBytes();
-        }
-        return totalBytes;
+    @Override
+    protected long estimateItemSize(Event item) {
+        return item.getSizeInBytes();
     }
 }