Add support for canceling Jobs.

app/src/main/java/org/thoughtcrime/securesms/jobmanager/Job.java

@@ -32,8 +32,10 @@ public abstract class Job {
 
   private final Parameters parameters;
 
-  private int    runAttempt;
-  private long   nextRunAttemptTime;
+  private int  runAttempt;
+  private long nextRunAttemptTime;
+
+  private volatile boolean canceled;
 
   protected Context context;
 
@@ -75,12 +77,27 @@ public abstract class Job {
     this.nextRunAttemptTime = nextRunAttemptTime;
   }
 
+  /** Should only be invoked by {@link JobController} */
+  final void cancel() {
+    this.canceled = true;
+  }
+
   @WorkerThread
   final void onSubmit() {
     Log.i(TAG, JobLogger.format(this, "onSubmit()"));
     onAdded();
   }
 
+  /**
+   * @return True if your job has been marked as canceled while it was running, otherwise false.
+   *         If a job sees that it has been canceled, it should make a best-effort attempt at
+   *         stopping it's work. This job will have {@link #onFailure()} called after {@link #run()}
+   *         has finished.
+   */
+  public final boolean isCanceled() {
+    return canceled;
+  }
+
   /**
    * Called when the job is first submitted to the {@link JobManager}.
    */
@@ -112,10 +129,10 @@ public abstract class Job {
   public abstract @NonNull Result run();
 
   /**
-   * Called when your job has completely failed.
+   * Called when your job has completely failed and will not be run again.
    */
   @WorkerThread
-  public abstract void onCanceled();
+  public abstract void onFailure();
 
   public interface Factory<T extends Job> {
     @NonNull T create(@NonNull Parameters parameters, @NonNull Data data);

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobController.java

@@ -17,11 +17,12 @@ import org.thoughtcrime.securesms.util.Debouncer;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
-import java.util.UUID;
 
 /**
  * Manages the queue of jobs. This is the only class that should write to {@link JobStorage} to
@@ -40,7 +41,7 @@ class JobController {
   private final Scheduler              scheduler;
   private final Debouncer              debouncer;
   private final Callback               callback;
-  private final Set<String>            runningJobs;
+  private final Map<String, Job>       runningJobs;
 
   JobController(@NonNull Application application,
                 @NonNull JobStorage jobStorage,
@@ -61,7 +62,7 @@ class JobController {
     this.scheduler              = scheduler;
     this.debouncer              = debouncer;
     this.callback               = callback;
-    this.runningJobs            = new HashSet<>();
+    this.runningJobs            = new HashMap<>();
   }
 
   @WorkerThread
@@ -96,6 +97,29 @@ class JobController {
     notifyAll();
   }
 
+  @WorkerThread
+  synchronized void cancelJob(@NonNull String id) {
+    Job runningJob = runningJobs.get(id);
+
+    if (runningJob != null) {
+      Log.w(TAG, JobLogger.format(runningJob, "Canceling while running."));
+      runningJob.cancel();
+    } else {
+      JobSpec jobSpec = jobStorage.getJobSpec(id);
+
+      if (jobSpec != null) {
+        Job job = createJob(jobSpec, jobStorage.getConstraintSpecs(id));
+        Log.w(TAG, JobLogger.format(job, "Canceling while inactive."));
+        Log.w(TAG, JobLogger.format(job, "Job failed."));
+
+        job.onFailure();
+        onFailure(job);
+      } else {
+        Log.w(TAG, "Tried to cancel JOB::" + id + ", but it could not be found.");
+      }
+    }
+  }
+
   @WorkerThread
   synchronized void onRetry(@NonNull Job job) {
     int    nextRunAttempt     = job.getRunAttempt() + 1;
@@ -177,7 +201,7 @@ class JobController {
       }
 
       jobStorage.updateJobRunningState(job.getId(), true);
-      runningJobs.add(job.getId());
+      runningJobs.put(job.getId(), job);
       jobTracker.onStateChange(job.getId(), JobTracker.JobState.RUNNING);
 
       return job;
@@ -333,7 +357,7 @@ class JobController {
 
       return job;
     } catch (RuntimeException e) {
-      Log.e(TAG, "Failed to instantiate job! Failing it and its dependencies without calling Job#onCanceled. Crash imminent.");
+      Log.e(TAG, "Failed to instantiate job! Failing it and its dependencies without calling Job#onFailure. Crash imminent.");
 
       List<String> failIds = Stream.of(jobStorage.getDependencySpecsThatDependOnJob(jobSpec.getId()))
                                    .map(DependencySpec::getJobId)

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobManager.java

@@ -112,7 +112,6 @@ public class JobManager implements ConstraintObserver.Notifier {
     jobTracker.removeListener(listener);
   }
 
-
   /**
    * Enqueues a single job to be run.
    */
@@ -136,9 +135,22 @@ public class JobManager implements ConstraintObserver.Notifier {
     return new Chain(this, jobs);
   }
 
+  /**
+   * Attempts to cancel a job. This is best-effort and may not actually prevent a job from
+   * completing if it was already running. If this job is running, this can only stop jobs that
+   * bother to check {@link Job#isCanceled()}.
+   *
+   * When a job is canceled, {@link Job#onFailure()} will be triggered at the earliest possible
+   * moment. Just like a normal failure, all later jobs in the same chain will also be failed.
+   */
+  public void cancel(@NonNull String id) {
+    executor.execute(() -> jobController.cancelJob(id));
+  }
+
   /**
    * Retrieves a string representing the state of the job queue. Intended for debugging.
    */
+  @WorkerThread
   public @NonNull String getDebugInfo() {
     Future<String> result = executor.submit(jobController::getDebugInfo);
     try {

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobRunner.java

@@ -54,8 +54,8 @@ class JobRunner extends Thread {
         job.onRetry();
       } else if (result.isFailure()) {
         List<Job> dependents = jobController.onFailure(job);
-        job.onCanceled();
-        Stream.of(dependents).forEach(Job::onCanceled);
+        job.onFailure();
+        Stream.of(dependents).forEach(Job::onFailure);
 
         if (result.getException() != null) {
           throw result.getException();
@@ -80,6 +80,11 @@ class JobRunner extends Thread {
     try {
       wakeLock = WakeLockUtil.acquire(application, PowerManager.PARTIAL_WAKE_LOCK, WAKE_LOCK_TIMEOUT, job.getId());
       result = job.run();
+
+      if (job.isCanceled()) {
+        Log.w(TAG, JobLogger.format(job, String.valueOf(id), "Failing because the job was canceled."));
+        result = Job.Result.failure();
+      }
     } catch (Exception e) {
       Log.w(TAG, JobLogger.format(job, String.valueOf(id), "Failing due to an unexpected exception."), e);
       return Job.Result.failure();