job status implementation #18

Author: yigit

jobqueue/src/com/path/android/jobqueue/JobManager.java

@@ -304,14 +304,41 @@ private void reAddJob(JobHolder jobHolder) {
     }
 
     /**
-     * returns the current status of a {@link Job}
+     * Returns the current status of a {@link Job}.
+     * <p>
+     *     You should not call this method on UI thread because it may make a db request.
+     * </p>
      * @param id the ID, returned by the addJob method
      * @param isPersistent Jobs are added to different queues depending on if they are persistent or not. This is necessary
      *                     because each queue has independent id sets.
      * @return
      */
     public JobStatus getJobStatus(long id, boolean isPersistent) {
-        return null;
+        if(jobConsumerExecutor.isRunning(id, isPersistent)) {
+            return JobStatus.RUNNING;
+        }
+        JobHolder holder;
+        if(isPersistent) {
+            synchronized (persistentJobQueue) {
+                holder = persistentJobQueue.findJobById(id);
+            }
+        } else {
+            synchronized (nonPersistentJobQueue) {
+                holder = nonPersistentJobQueue.findJobById(id);
+            }
+        }
+        if(holder == null) {
+            return JobStatus.UNKNOWN;
+        }
+        boolean network = hasNetwork();
+        if(holder.requiresNetwork() && !network) {
+            return JobStatus.WAITING_NOT_READY;
+        }
+        if(holder.getDelayUntilNs() > System.nanoTime()) {
+            return JobStatus.WAITING_NOT_READY;
+        }
+
+        return JobStatus.WAITING_READY;
     }
 
     private void removeJob(JobHolder jobHolder) {

jobqueue/src/com/path/android/jobqueue/JobStatus.java

@@ -9,6 +9,7 @@ public enum JobStatus {
      * As of v 1.1, this might be:
      * <ul>
      *     <li>Job requires network but there is no available network connection</li>
+     *     <li>Job is delayed. We are waiting for the time to pass</li>
      * </ul>
      */
     WAITING_NOT_READY,

jobqueue/src/com/path/android/jobqueue/executor/JobConsumerExecutor.java

@@ -6,6 +6,7 @@
 import com.path.android.jobqueue.config.Configuration;
 import com.path.android.jobqueue.log.JqLog;
 
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -21,7 +22,8 @@ public class JobConsumerExecutor {
     private final Contract contract;
     private final int keepAliveSeconds;
     private final AtomicInteger activeConsumerCount = new AtomicInteger(0);
-    private final AtomicInteger runningJobCount = new AtomicInteger(0);
+    // key : id + (isPersistent)
+    private final ConcurrentHashMap<String, JobHolder> runningJobHolders;
 
 
     public JobConsumerExecutor(Configuration config, Contract contract) {
@@ -31,6 +33,7 @@ public JobConsumerExecutor(Configuration config, Contract contract) {
         this.keepAliveSeconds = config.getConsumerKeepAlive();
         this.contract = contract;
         threadGroup = new ThreadGroup("JobConsumers");
+        runningJobHolders = new ConcurrentHashMap<String, JobHolder>();
     }
 
     /**
@@ -92,17 +95,43 @@ private boolean isAboveLoadFactor(boolean inConsumerThread) {
             int consumerCnt = activeConsumerCount.intValue() - (inConsumerThread ? 1 : 0);
             boolean res =
                     consumerCnt < minConsumerSize ||
-                    consumerCnt * loadFactor < contract.countRemainingReadyJobs() + runningJobCount.get();
+                    consumerCnt * loadFactor < contract.countRemainingReadyJobs() + runningJobHolders.size();
             if(JqLog.isDebugEnabled()) {
                 JqLog.d("%s: load factor check. %s = (%d < %d)|| (%d * %d < %d + %d). consumer thread: %s", Thread.currentThread().getName(), res,
                         consumerCnt, minConsumerSize,
-                        consumerCnt, loadFactor, contract.countRemainingReadyJobs(), runningJobCount.get(), inConsumerThread);
+                        consumerCnt, loadFactor, contract.countRemainingReadyJobs(), runningJobHolders.size(), inConsumerThread);
             }
             return res;
         }
 
     }
 
+    private void onBeforeRun(JobHolder jobHolder) {
+        runningJobHolders.put(createrunningJobHolderKey(jobHolder), jobHolder);
+    }
+
+    private void onAfterRun(JobHolder jobHolder) {
+        runningJobHolders.remove(createrunningJobHolderKey(jobHolder));
+    }
+
+    private String createrunningJobHolderKey(JobHolder jobHolder) {
+        return createrunningJobHolderKey(jobHolder.getId(), jobHolder.getBaseJob().isPersistent());
+    }
+
+    private String createrunningJobHolderKey(long id, boolean isPersistent) {
+        return id + "_" + (isPersistent ? "t" : "f");
+    }
+
+    /**
+     * returns true if job is currently handled by one of the executor threads
+     * @param id id of the job
+     * @param persistent boolean flag to distinguish id conflicts
+     * @return true if job is currently handled here
+     */
+    public boolean isRunning(long id, boolean persistent) {
+        return runningJobHolders.containsKey(createrunningJobHolderKey(id, persistent));
+    }
+
     /**
      * contract between the {@link JobManager} and {@link JobConsumerExecutor}
      */
@@ -168,13 +197,13 @@ public void run() {
                     do {
                         nextJob = contract.isRunning() ? contract.getNextJob(executor.keepAliveSeconds, TimeUnit.SECONDS) : null;
                         if (nextJob != null) {
-                            executor.runningJobCount.incrementAndGet();
+                            executor.onBeforeRun(nextJob);
                             if (nextJob.safeRun(nextJob.getRunCount())) {
                                 contract.removeJob(nextJob);
                             } else {
                                 contract.insertOrReplace(nextJob);
                             }
-                            executor.runningJobCount.decrementAndGet();
+                            executor.onAfterRun(nextJob);
                         }
                     } while (nextJob != null);
                 } finally {

jobqueue/test/com/path/android/jobqueue/test/jobmanager/JobStatusTest.java

@@ -91,5 +91,29 @@ public void testJobStatus() throws InterruptedException {
             //after all jobs finish, state should be unknown
             assertThat("all jobs finished, states should be unknown", jobManager.getJobStatus(ids[i], jobs[i].isPersistent()), is(JobStatus.UNKNOWN));
         }
+        final long SHORT_SLEEP = 1000;
+        Job[] delayedJobs = new Job[]{
+                new DummyJob(new Params(0).delayInMs(SHORT_SLEEP)),
+                new DummyJob(new Params(0).delayInMs(SHORT_SLEEP).persist()),
+                new DummyJob(new Params(0).delayInMs(SHORT_SLEEP * 10)),
+                new DummyJob(new Params(0).delayInMs(SHORT_SLEEP * 10).persist())};
+        long[] delayedIds = new long[delayedJobs.length];
+        for(int i = 0; i < delayedJobs.length; i ++) {
+            delayedIds[i] = jobManager.addJob(delayedJobs[i]);
+        }
+
+        for(int i = 0; i < delayedJobs.length; i ++) {
+            assertThat("delayed job(" + i + ") should receive not ready status", jobManager.getJobStatus(delayedIds[i], delayedJobs[i].isPersistent()), is(JobStatus.WAITING_NOT_READY));
+        }
+        jobManager.stop();
+        //sleep
+        Thread.sleep(SHORT_SLEEP * 2);
+        for(int i = 0; i < delayedJobs.length; i ++) {
+            if(delayedJobs[i].getDelayInMs() == SHORT_SLEEP) {
+                assertThat("when enough time passes, delayed jobs should move to ready state", jobManager.getJobStatus(delayedIds[i], delayedJobs[i].isPersistent()), is(JobStatus.WAITING_READY));
+            } else {
+                assertThat("delayed job should receive not ready status until their time comes", jobManager.getJobStatus(delayedIds[i], delayedJobs[i].isPersistent()), is(JobStatus.WAITING_NOT_READY));
+            }
+        }
     }
 }