[SDK-707] Preserve legacy request execution behavior

Author: franco-zalamena-iterable

CHANGELOG.md

@@ -5,7 +5,7 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 ## [Unreleased]
 
 ### Fixed
-- Push token registration, disable operations, Iterable deep-link redirects, and API requests now use dedicated SDK-owned executors instead of Android's process-wide `AsyncTask` queues. Push, deep-link, and offline operations preserve ordering in isolated serial lanes, while ordinary online API requests retain concurrent execution. Slow work in one lane no longer delays unrelated SDK operations, and client callbacks and attribution updates continue on the main thread.
+- Push token registration, disable operations, Iterable deep-link redirects, and API requests now use dedicated SDK-owned executors instead of Android's process-wide `AsyncTask` queues. Push and deep-link work retain isolated serial lanes. Ordinary online API requests remain concurrent, while their automatic retries and offline immediate requests retain serial ordering; stored offline requests retain their own FIFO flow. Slow work in one flow no longer delays unrelated SDK operations, and client callbacks and attribution updates continue on the main thread.
 
 ## [3.11.0]
 ### Added

iterableapi/src/main/java/com/iterable/iterableapi/IterableApiClient.java

@@ -26,6 +26,7 @@ class IterableApiClient {
     // A newer push action invalidates retries from earlier registration or disable requests.
     private final AtomicLong pushRegistrationRequestGeneration = new AtomicLong();
     private RequestProcessor requestProcessor;
+    private OfflineRequestProcessor offlineRequestProcessor;
 
     interface AuthProvider {
         @Nullable
@@ -59,32 +60,50 @@ interface AuthProvider {
                 new OnlineRequestProcessor(requestDispatchers.push());
     }
 
-    private RequestProcessor getRequestProcessor() {
+    private synchronized RequestProcessor getRequestProcessor() {
         if (requestProcessor == null) {
             requestProcessor = new OnlineRequestProcessor(requestDispatchers.online());
         }
         return requestProcessor;
     }
 
-    void setOfflineProcessingEnabled(boolean offlineMode) {
-        if (offlineMode && this.requestProcessor instanceof OfflineRequestProcessor) {
-            return;
-        }
-        if (!offlineMode && this.requestProcessor instanceof OnlineRequestProcessor) {
+    synchronized void setOfflineProcessingEnabled(boolean offlineMode) {
+        if (!offlineMode
+                && offlineRequestProcessor == null
+                && !hasPendingOfflineRequests()) {
+            if (!(requestProcessor instanceof OnlineRequestProcessor)) {
+                requestProcessor = new OnlineRequestProcessor(requestDispatchers.online());
+            }
             return;
         }
 
-        if (this.requestProcessor instanceof OfflineRequestProcessor) {
-            ((OfflineRequestProcessor) this.requestProcessor).dispose();
+        // Once needed, keep one offline processor alive while new requests use the
+        // online processor. Its runner must finish requests queued before offline
+        // processing was disabled or restored from a previous process.
+        OfflineRequestProcessor persistentOfflineRequestProcessor =
+                getOfflineRequestProcessor();
+        if (offlineMode) {
+            requestProcessor = persistentOfflineRequestProcessor;
+        } else if (!(requestProcessor instanceof OnlineRequestProcessor)) {
+            requestProcessor = new OnlineRequestProcessor(requestDispatchers.online());
         }
+    }
 
-        this.requestProcessor = offlineMode
-                ? new OfflineRequestProcessor(
+    private OfflineRequestProcessor getOfflineRequestProcessor() {
+        if (offlineRequestProcessor == null) {
+            offlineRequestProcessor = new OfflineRequestProcessor(
                         authProvider.getContext(),
-                        requestDispatchers.online(),
-                        requestDispatchers.offline()
-                )
-                : new OnlineRequestProcessor(requestDispatchers.online());
+                        requestDispatchers.offlineImmediate(),
+                        requestDispatchers.offlineStored()
+            );
+        }
+        return offlineRequestProcessor;
+    }
+
+    private boolean hasPendingOfflineRequests() {
+        return IterableTaskStorage
+                .sharedInstance(authProvider.getContext())
+                .hasPendingTasks();
     }
 
     void getRemoteConfiguration(IterableHelper.IterableActionHandler actionHandler) {
@@ -832,11 +851,30 @@ void sendGetRequest(@NonNull String resourcePath, @NonNull JSONObject json, @Non
         getRequestProcessor().processGetRequest(authProvider.getApiKey(), resourcePath, json, authProvider.getAuthToken(), onSuccess, onFailure);
     }
 
-    void onLogout() {
-        getRequestProcessor().onLogout(authProvider.getContext());
+    synchronized void onLogout() {
+        Context context = authProvider.getContext();
+        RequestProcessor activeRequestProcessor = getRequestProcessor();
+        if (offlineRequestProcessor != null) {
+            offlineRequestProcessor.onLogout(context);
+        } else {
+            IterableTaskStorage.sharedInstance(context).deleteAllTasks();
+        }
+        if (activeRequestProcessor != offlineRequestProcessor) {
+            activeRequestProcessor.onLogout(context);
+        }
         authProvider.resetAuth();
     }
 
+    synchronized void dispose() {
+        if (offlineRequestProcessor != null) {
+            if (requestProcessor == offlineRequestProcessor) {
+                requestProcessor = null;
+            }
+            offlineRequestProcessor.dispose();
+            offlineRequestProcessor = null;
+        }
+    }
+
     void mergeUser(String sourceEmail, String sourceUserId, String destinationEmail, String destinationUserId, @Nullable IterableHelper.SuccessHandler successHandler, @Nullable IterableHelper.FailureHandler failureHandler) {
         JSONObject requestJson = new JSONObject();
         try {

iterableapi/src/main/java/com/iterable/iterableapi/IterableExecutors.java

@@ -7,6 +7,7 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -15,19 +16,32 @@ final class IterableExecutors {
     // HttpURLConnection is blocking I/O, so ordinary API work uses a bounded
     // multi-thread pool rather than the serial executors used for ordered work.
     static final int REQUEST_THREAD_COUNT = 8;
-    // Matches the historical AsyncTask queue bound without inheriting its
+    // Bounds the primary pool without inheriting AsyncTask's
     // platform-version-dependent thread-pool behavior.
     static final int REQUEST_QUEUE_CAPACITY = 128;
     private static final long REQUEST_THREAD_KEEP_ALIVE_SECONDS = 30;
+    private static final int REQUEST_OVERFLOW_THREAD_COUNT = 5;
+    private static final long REQUEST_OVERFLOW_KEEP_ALIVE_SECONDS = 3;
     private static final AtomicInteger REQUEST_THREAD_ID = new AtomicInteger();
+    private static final AtomicInteger REQUEST_OVERFLOW_THREAD_ID = new AtomicInteger();
     private static final Executor PUSH_EXECUTOR =
             newSingleThreadExecutor("IterablePushExecutor");
     private static final Executor DEEP_LINK_EXECUTOR =
             newSingleThreadExecutor("IterableDeepLinkExecutor");
-    private static final Executor OFFLINE_EXECUTOR =
-            newSingleThreadExecutor("IterableOfflineExecutor");
+    private static final Executor SERIAL_REQUEST_EXECUTOR =
+            newSingleThreadExecutor("IterableSerialRequestExecutor");
+    private static final Executor OFFLINE_STORED_EXECUTOR =
+            newSingleThreadExecutor("IterableOfflineStoredExecutor");
+    // AsyncTask's concurrent executor used a backup queue instead of dropping
+    // work when its primary pool was saturated. Preserve that delivery behavior.
+    private static final Executor REQUEST_OVERFLOW_EXECUTOR =
+            newRequestOverflowExecutor(REQUEST_OVERFLOW_THREAD_COUNT);
     private static final Executor REQUEST_EXECUTOR =
-            newRequestExecutor(REQUEST_THREAD_COUNT, REQUEST_QUEUE_CAPACITY);
+            newRequestExecutor(
+                    REQUEST_THREAD_COUNT,
+                    REQUEST_QUEUE_CAPACITY,
+                    REQUEST_OVERFLOW_EXECUTOR
+            );
     private static final Executor MAIN_EXECUTOR =
             runnable -> new Handler(Looper.getMainLooper()).post(runnable);
 
@@ -46,8 +60,16 @@ static Executor request() {
         return REQUEST_EXECUTOR;
     }
 
-    static Executor offline() {
-        return OFFLINE_EXECUTOR;
+    static Executor offlineImmediate() {
+        return SERIAL_REQUEST_EXECUTOR;
+    }
+
+    static Executor requestRetry() {
+        return SERIAL_REQUEST_EXECUTOR;
+    }
+
+    static Executor offlineStored() {
+        return OFFLINE_STORED_EXECUTOR;
     }
 
     static Executor main() {
@@ -61,6 +83,14 @@ private static Executor newSingleThreadExecutor(String threadName) {
     }
 
     static ThreadPoolExecutor newRequestExecutor(int threadCount, int queueCapacity) {
+        return newRequestExecutor(threadCount, queueCapacity, REQUEST_OVERFLOW_EXECUTOR);
+    }
+
+    static ThreadPoolExecutor newRequestExecutor(
+            int threadCount,
+            int queueCapacity,
+            Executor overflowExecutor
+    ) {
         ThreadPoolExecutor executor = new ThreadPoolExecutor(
                 threadCount,
                 threadCount,
@@ -71,7 +101,24 @@ static ThreadPoolExecutor newRequestExecutor(int threadCount, int queueCapacity)
                         runnable,
                         "IterableRequestExecutor-" + REQUEST_THREAD_ID.incrementAndGet()
                 ),
-                new ThreadPoolExecutor.AbortPolicy()
+                (runnable, ignored) -> overflowExecutor.execute(runnable)
+        );
+        executor.allowCoreThreadTimeOut(true);
+        return executor;
+    }
+
+    static ThreadPoolExecutor newRequestOverflowExecutor(int threadCount) {
+        ThreadPoolExecutor executor = new ThreadPoolExecutor(
+                threadCount,
+                threadCount,
+                REQUEST_OVERFLOW_KEEP_ALIVE_SECONDS,
+                TimeUnit.SECONDS,
+                new LinkedBlockingQueue<>(),
+                runnable -> newThread(
+                        runnable,
+                        "IterableRequestOverflowExecutor-"
+                                + REQUEST_OVERFLOW_THREAD_ID.incrementAndGet()
+                )
         );
         executor.allowCoreThreadTimeOut(true);
         return executor;