Update AndroidJobStrategy to cancel requests

Author: adimiz1

core/src/androidTest/java/com/cloudinary/android/AndroidJobStrategyTest.java

@@ -3,15 +3,21 @@
 
 import androidx.test.internal.runner.junit4.AndroidJUnit4ClassRunner;
 import androidx.test.internal.util.ReflectionUtil;
+import androidx.test.platform.app.InstrumentationRegistry;
 import androidx.work.BackoffPolicy;
 import androidx.work.Constraints;
 import androidx.work.OneTimeWorkRequest;
 import androidx.work.WorkManager;
 import androidx.work.WorkRequest;
 import androidx.work.impl.model.WorkSpec;
 
+import com.cloudinary.Cloudinary;
+import com.cloudinary.android.callback.ErrorInfo;
+import com.cloudinary.android.callback.UploadCallback;
 import com.cloudinary.android.payload.FilePayload;
+import com.cloudinary.utils.ObjectUtils;
 
+import org.cloudinary.json.JSONObject;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -20,10 +26,13 @@
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.lang.reflect.ParameterizedType;
+import java.util.Map;
 
 @RunWith(AndroidJUnit4ClassRunner.class)
 public class AndroidJobStrategyTest extends AbstractTest {
 
+    int success = 0;
+    int errors = 0;
     private File payloadFile;
 
     @Override
@@ -55,4 +64,39 @@ public void testAdapter() throws InterruptedException, IOException, NoSuchFieldE
         Assert.assertEquals(BackoffPolicy.LINEAR, adapted.getWorkSpec().backoffPolicy);
 
     }
+    @Test
+    public void testCancelRequest() throws InterruptedException, IOException, NoSuchFieldException, IllegalAccessException {
+        FilePayload payload = buildPayload();
+        String requestId = MediaManager.get().upload(payload)
+                .unsigned(TEST_PRESET).callback(new UploadCallback() {
+                    @Override
+                    public void onStart(String requestId) {
+                    }
+
+                    @Override
+                    public void onProgress(String requestId, long bytes, long totalBytes) {
+                    }
+
+                    @Override
+                    public void onSuccess(String requestId, Map resultData) {
+                        success++;
+                    }
+
+                    @Override
+                    public void onError(String requestId, ErrorInfo error) {
+                        errors++;
+                    }
+
+                    @Override
+                    public void onReschedule(String requestId, ErrorInfo error) {
+
+                    }
+                })
+                .dispatch();
+        Thread.sleep(1000);
+        MediaManager.get().cancelRequest(requestId);
+        Thread.sleep(3000);
+        Assert.assertTrue(success == 0);
+        Assert.assertTrue(errors == 1);
+    }
 }

core/src/main/java/com/cloudinary/android/AndroidJobStrategy.java

@@ -2,7 +2,6 @@
 
 import android.content.Context;
 import android.os.Bundle;
-
 import androidx.annotation.NonNull;
 import androidx.work.BackoffPolicy;
 import androidx.work.Constraints;
@@ -23,13 +22,19 @@
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.ObjectInputStream;
+import java.lang.ref.WeakReference;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
 public class AndroidJobStrategy implements BackgroundRequestStrategy {
 
     private static final String JOB_TAG = "CLD";
 
+    private static final Map<String, WeakReference<Thread>> threads = new ConcurrentHashMap<>();
+    private static final Object threadsMapLockObject = new Object();
+
     private Context context;
 
     public static OneTimeWorkRequest adapt(UploadRequest<?> request, File payloadFile) {
@@ -43,7 +48,7 @@ public static OneTimeWorkRequest adapt(UploadRequest<?> request, File payloadFil
 
         Data inputData = request.buildPayload(payloadFile);
 
-        return new OneTimeWorkRequest.Builder(UploadJob.class).setBackoffCriteria(adaptBackoffPolicy(policy.getBackoffPolicy()), policy.getBackoffMillis(), TimeUnit.MILLISECONDS).setInputData(inputData).setConstraints(constraints).addTag(JOB_TAG).build();
+        return new OneTimeWorkRequest.Builder(UploadJob.class).setBackoffCriteria(adaptBackoffPolicy(policy.getBackoffPolicy()), policy.getBackoffMillis(), TimeUnit.MILLISECONDS).setInputData(inputData).setConstraints(constraints).addTag(request.getRequestId()).build();
     }
 
     private static BackoffPolicy adaptBackoffPolicy(UploadPolicy.BackoffPolicy backoffPolicy) {
@@ -99,15 +104,48 @@ public void executeRequestsNow(int howMany) {
     @Override
     public boolean cancelRequest(String requestId) {
         Operation operation = WorkManager.getInstance(context).cancelAllWorkByTag(requestId);
+        killThread(requestId);
         return operation.getResult().isCancelled();
     }
 
     @Override
     public int cancelAllRequests() {
         WorkManager.getInstance(context).cancelAllWork();
+        killAllThreads();
         return 0;
     }
 
+    private void killThread(String requestId) {
+        synchronized (threadsMapLockObject) {
+            WeakReference<Thread> ref = threads.remove(requestId);
+            if (ref != null) {
+                Thread thread = ref.get();
+                if (thread != null) {
+                    thread.interrupt();
+                }
+
+                ref.clear();
+            }
+        }
+    }
+
+    private void killAllThreads() {
+        synchronized (threadsMapLockObject) {
+            for (String requestId : threads.keySet()) {
+                WeakReference<Thread> ref = threads.get(requestId);
+                Thread thread = ref.get();
+
+                if (thread != null) {
+                    thread.interrupt();
+                }
+
+                ref.clear();
+            }
+
+            threads.clear();
+        }
+    }
+
     @Override
     public int getPendingImmediateJobsCount() {
         return getJobCountByState(WorkInfo.State.ENQUEUED);
@@ -137,36 +175,62 @@ public static final class UploadJob extends Worker {
         private final Context context;
         private final WorkerParameters workParams;
 
+        private String requestId;
+
         public UploadJob(@NonNull Context context, @NonNull WorkerParameters workerParams) {
             super(context, workerParams);
             this.context = context;
             this.workParams = workerParams;
         }
 
+        @Override
+        public void onStopped() {
+            super.onStopped();
+            unregisterThread(requestId);
+        }
+
+        private void registerThread(String requestId, Thread thread) {
+            synchronized (threadsMapLockObject) {
+                threads.put(requestId, new WeakReference<>(thread));
+            }
+        }
+
+        private void unregisterThread(String requestId) {
+            synchronized (threadsMapLockObject) {
+                if(requestId != null) {
+                    WeakReference<Thread> removed = threads.remove(requestId);
+                    if (removed != null) {
+                        removed.clear();
+                    }
+                }
+            }
+        }
+
         @NonNull
         @Override
         public Result doWork() {
-            // Removed Wakelock logic as it causes RuntimeException ("WakeLock under-locked")
 
             // Prepare extract payload data from temporary file.
             String payloadFilePath = workParams.getInputData().getString(UploadRequest.PayloadData.KEY);
             if (payloadFilePath == null) {
                 // NO Payload input file created prior to request.
                 return Result.failure();
             }
+
             File payloadFile = new File(payloadFilePath);
             try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(payloadFile))) {
                 UploadRequest.PayloadData payloadData = (UploadRequest.PayloadData) ois.readObject();
                 AndroidJobStrategy.AndroidJobRequestParams jobInputData = new AndroidJobStrategy.AndroidJobRequestParams(payloadData);
-
-                // call the generic processor:
-                UploadStatus result = MediaManager.get().processRequest(context, jobInputData);
+                requestId = payloadData.getRequestId();
+                registerThread(requestId, Thread.currentThread());
+                UploadStatus result = MediaManager.get().processRequest(context, jobInputData); // Replace this with your actual upload logic
                 return adaptResult(result);
-
-            } catch (NullPointerException | IOException | ClassNotFoundException e) {
+            } catch (NullPointerException | IOException | ClassNotFoundException e ) {
                 // Unable to deserialize payload data from file.
                 e.printStackTrace();
                 return Result.failure();
+            } finally {
+                unregisterThread(requestId);
             }
         }