runningJobIds = runningJobGroups.getSafe();
+ synchronized (nonPersistentJobQueue) {
+ jobHolder = nonPersistentJobQueue.nextJobAndIncRunCount(haveNetwork, runningJobIds);
}
- }
- if(jobHolder != null) {
- //wait for onAdded locks
- if(persistent) {
- waitForOnAddedLock(persistentOnAddedLocks, jobHolder.getId());
- } else {
- waitForOnAddedLock(nonPersistentOnAddedLocks, jobHolder.getId());
+ if (jobHolder == null) {
+ //go to disk, there aren't any non-persistent jobs
+ synchronized (persistentJobQueue) {
+ jobHolder = persistentJobQueue.nextJobAndIncRunCount(haveNetwork, runningJobIds);
+ persistent = true;
+ }
+ }
+ if(jobHolder == null) {
+ return null;
+ }
+ if(persistent && dependencyInjector != null) {
+ dependencyInjector.inject(jobHolder.getBaseJob());
+ }
+ if(jobHolder.getGroupId() != null) {
+ runningJobGroups.add(jobHolder.getGroupId());
}
}
- if(persistent && jobHolder != null && dependencyInjector != null) {
- dependencyInjector.inject(jobHolder.getBaseJob());
- }
- if(jobHolder != null && jobHolder.getGroupId() != null) {
- runningJobGroups.add(jobHolder.getGroupId());
+
+ //wait for onAdded locks. wait for locks after job is selected so that we minimize the lock
+ if(persistent) {
+ waitForOnAddedLock(persistentOnAddedLocks, jobHolder.getId());
+ } else {
+ waitForOnAddedLock(nonPersistentOnAddedLocks, jobHolder.getId());
}
+
return jobHolder;
}
@@ -299,6 +311,48 @@ private void reAddJob(JobHolder jobHolder) {
}
}
+ /**
+ * Returns the current status of a {@link Job}.
+ *
+ * You should not call this method on the UI thread because it may make a db request.
+ *
+ *
+ * This is not a very fast call so try not to make it unless necessary. Consider using events if you need to be
+ * informed about a job's lifecycle.
+ *
+ * @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) {
+ 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) {
if (jobHolder.getBaseJob().isPersistent()) {
synchronized (persistentJobQueue) {
@@ -497,12 +551,24 @@ public void run() {
*/
@Deprecated
public void addJobInBackground(final int priority, final long delay, final BaseJob baseJob) {
+ addJobInBackground(priority, delay, baseJob, null);
+ }
+
+ protected void addJobInBackground(final int priority, final long delay, final BaseJob baseJob,
+ /*nullable*/final AsyncAddCallback callback) {
final long callTime = System.nanoTime();
timedExecutor.execute(new Runnable() {
@Override
public void run() {
- final long runDelay = (System.nanoTime() - callTime) / NS_PER_MS;
- addJob(priority, Math.max(0, delay - runDelay), baseJob);
+ try {
+ final long runDelay = (System.nanoTime() - callTime) / NS_PER_MS;
+ long id = addJob(priority, Math.max(0, delay - runDelay), baseJob);
+ if(callback != null) {
+ callback.onAdded(id);
+ }
+ } catch (Throwable t) {
+ JqLog.e(t, "addJobInBackground received an exception. job class: %s", baseJob.getClass().getSimpleName() );
+ }
}
});
}
diff --git a/jobqueue/src/com/path/android/jobqueue/JobQueue.java b/jobqueue/src/com/path/android/jobqueue/JobQueue.java
index 0755505..a149a00 100644
--- a/jobqueue/src/com/path/android/jobqueue/JobQueue.java
+++ b/jobqueue/src/com/path/android/jobqueue/JobQueue.java
@@ -72,4 +72,11 @@ public interface JobQueue {
*/
void clear();
+ /**
+ * returns the job with the given id if it exists in the queue
+ * @param id id of the job, returned by insert method
+ * @return JobHolder with the given id or null if it does not exists
+ */
+ JobHolder findJobById(long id);
+
}
diff --git a/jobqueue/src/com/path/android/jobqueue/JobStatus.java b/jobqueue/src/com/path/android/jobqueue/JobStatus.java
new file mode 100644
index 0000000..b687cc3
--- /dev/null
+++ b/jobqueue/src/com/path/android/jobqueue/JobStatus.java
@@ -0,0 +1,36 @@
+package com.path.android.jobqueue;
+
+/**
+ * Identifies the current status of a job if it is in the queue
+ */
+public enum JobStatus {
+ /**
+ * Job is in the queue but cannot run yet.
+ * As of v 1.1, this might be:
+ *
+ * - Job requires network but there is no available network connection
+ * - Job is delayed. We are waiting for the time to pass
+ *
+ */
+ WAITING_NOT_READY,
+ /**
+ * Job is in the queue, ready to be run. Waiting for an available consumer.
+ */
+ WAITING_READY,
+ /**
+ * Job is being executed by one of the runners.
+ */
+ RUNNING,
+ /**
+ * Job is not known by job queue.
+ * This might be:
+ *
+ * - Invalid ID
+ * - Job has been completed
+ * - Job has failed
+ * - Job has just been added, about to be delivered into a queue
+ *
+ *
+ */
+ UNKNOWN
+}
diff --git a/jobqueue/src/com/path/android/jobqueue/Params.java b/jobqueue/src/com/path/android/jobqueue/Params.java
index c3e323a..1695d94 100644
--- a/jobqueue/src/com/path/android/jobqueue/Params.java
+++ b/jobqueue/src/com/path/android/jobqueue/Params.java
@@ -97,23 +97,23 @@ public Params setDelayMs(long delayMs) {
return this;
}
- boolean doesRequireNetwork() {
+ public boolean doesRequireNetwork() {
return requiresNetwork;
}
- String getGroupId() {
+ public String getGroupId() {
return groupId;
}
- boolean isPersistent() {
+ public boolean isPersistent() {
return persistent;
}
- int getPriority() {
+ public int getPriority() {
return priority;
}
- long getDelayMs() {
+ public long getDelayMs() {
return delayMs;
}
}
diff --git a/jobqueue/src/com/path/android/jobqueue/cachedQueue/CachedJobQueue.java b/jobqueue/src/com/path/android/jobqueue/cachedQueue/CachedJobQueue.java
index be3279a..c373c99 100644
--- a/jobqueue/src/com/path/android/jobqueue/cachedQueue/CachedJobQueue.java
+++ b/jobqueue/src/com/path/android/jobqueue/cachedQueue/CachedJobQueue.java
@@ -93,6 +93,11 @@ public void clear() {
delegate.clear();
}
+ @Override
+ public JobHolder findJobById(long id) {
+ return delegate.findJobById(id);
+ }
+
private static class Cache {
Integer count;
DelayUntil delayUntil;
diff --git a/jobqueue/src/com/path/android/jobqueue/executor/JobConsumerExecutor.java b/jobqueue/src/com/path/android/jobqueue/executor/JobConsumerExecutor.java
index 2334f62..139d92c 100644
--- a/jobqueue/src/com/path/android/jobqueue/executor/JobConsumerExecutor.java
+++ b/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 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();
}
/**
@@ -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 {
diff --git a/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/JobSet.java b/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/JobSet.java
index 2613245..b084a0e 100644
--- a/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/JobSet.java
+++ b/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/JobSet.java
@@ -11,6 +11,7 @@
public interface JobSet {
public JobHolder peek(Collection excludeGroupIds);
public JobHolder poll(Collection excludeGroupIds);
+ public JobHolder findById(long id);
public boolean offer(JobHolder holder);
public boolean remove(JobHolder holder);
public void clear();
diff --git a/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/MergedQueue.java b/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/MergedQueue.java
index f845afd..f32fd7c 100644
--- a/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/MergedQueue.java
+++ b/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/MergedQueue.java
@@ -110,31 +110,33 @@ public JobHolder poll(Collection excludeGroupIds) {
*/
@Override
public JobHolder peek(Collection excludeGroupIds) {
- JobHolder delayed = queue0.peek(excludeGroupIds);
- //if queue for this job has changed, re-add it and try peek from scratch
- if(delayed != null && decideQueue(delayed) != SetId.S0) {
- queue1.offer(delayed);
- queue0.remove(delayed);
- return peek(excludeGroupIds);
- }
- JobHolder nonDelayed = queue1.peek(excludeGroupIds);
- //if queue for this job has changed, re-add it and try peek from scratch
- if(nonDelayed != null && decideQueue(nonDelayed) != SetId.S1) {
- queue0.offer(nonDelayed);
- queue1.remove(nonDelayed);
- return peek(excludeGroupIds);
- }
- if(delayed == null) {
+ while (true) {
+ JobHolder delayed = queue0.peek(excludeGroupIds);
+ //if queue for this job has changed, re-add it and try peek from scratch
+ if(delayed != null && decideQueue(delayed) != SetId.S0) {
+ queue1.offer(delayed);
+ queue0.remove(delayed);
+ continue;//retry
+ }
+ JobHolder nonDelayed = queue1.peek(excludeGroupIds);
+ //if queue for this job has changed, re-add it and try peek from scratch
+ if(nonDelayed != null && decideQueue(nonDelayed) != SetId.S1) {
+ queue0.offer(nonDelayed);
+ queue1.remove(nonDelayed);
+ continue;//retry
+ }
+ if(delayed == null) {
+ return nonDelayed;
+ }
+ if(nonDelayed == null) {
+ return delayed;
+ }
+ int cmp = retrieveComparator.compare(delayed, nonDelayed);
+ if(cmp == -1) {
+ return delayed;
+ }
return nonDelayed;
}
- if(nonDelayed == null) {
- return delayed;
- }
- int cmp = retrieveComparator.compare(delayed, nonDelayed);
- if(cmp == -1) {
- return delayed;
- }
- return nonDelayed;
}
@@ -199,6 +201,19 @@ public CountWithGroupIdsResult countReadyJobs(SetId setId, Collection ex
}
}
+
+
+ /**
+ * Returns the JobHolder that has the given id
+ * @param id id job the job
+ * @return
+ */
+ @Override
+ public JobHolder findById(long id) {
+ JobHolder q0 = queue0.findById(id);
+ return q0 == null ? queue1.findById(id) : q0;
+ }
+
/**
* simple enum to identify queues
*/
diff --git a/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/NonPersistentJobSet.java b/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/NonPersistentJobSet.java
index c178453..aec9272 100644
--- a/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/NonPersistentJobSet.java
+++ b/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/NonPersistentJobSet.java
@@ -3,7 +3,13 @@
import com.path.android.jobqueue.JobHolder;
import com.path.android.jobqueue.log.JqLog;
-import java.util.*;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
/**
* This is the default implementation of JobSet.
@@ -11,13 +17,15 @@
* version
*/
public class NonPersistentJobSet implements JobSet {
- TreeSet set;
+ private final TreeSet set;
//groupId -> # of jobs in that group
- Map existingGroups;
+ private final Map existingGroups;
+ private final Map idCache;
public NonPersistentJobSet(Comparator comparator) {
this.set = new TreeSet(comparator);
this.existingGroups = new HashMap();
+ this.idCache = new HashMap();
}
private JobHolder safeFirst() {
@@ -62,6 +70,11 @@ public JobHolder poll(Collection excludeGroupIds) {
return peek;
}
+ @Override
+ public JobHolder findById(long id) {
+ return idCache.get(id);
+ }
+
@Override
public boolean offer(JobHolder holder) {
if(holder.getId() == null) {
@@ -73,9 +86,13 @@ public boolean offer(JobHolder holder) {
remove(holder);
result = set.add(holder);
}
- if(result && holder.getGroupId() != null) {
- incGroupCount(holder.getGroupId());
+ if(result) {
+ idCache.put(holder.getId(), holder);
+ if(holder.getGroupId() != null) {
+ incGroupCount(holder.getGroupId());
+ }
}
+
return result;
}
@@ -103,8 +120,11 @@ private void decGroupCount(String groupId) {
@Override
public boolean remove(JobHolder holder) {
boolean removed = set.remove(holder);
- if(removed && holder.getGroupId() != null) {
- decGroupCount(holder.getGroupId());
+ if(removed) {
+ idCache.remove(holder.getId());
+ if(holder.getGroupId() != null) {
+ decGroupCount(holder.getGroupId());
+ }
}
return removed;
}
@@ -115,6 +135,7 @@ public boolean remove(JobHolder holder) {
public void clear() {
set.clear();
existingGroups.clear();
+ idCache.clear();
}
@Override
diff --git a/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/NonPersistentPriorityQueue.java b/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/NonPersistentPriorityQueue.java
index 618d4da..627d09b 100644
--- a/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/NonPersistentPriorityQueue.java
+++ b/jobqueue/src/com/path/android/jobqueue/nonPersistentQueue/NonPersistentPriorityQueue.java
@@ -91,11 +91,22 @@ public Long getNextJobDelayUntilNs(boolean hasNetwork) {
return next == null ? null : next.getDelayUntilNs();
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public void clear() {
jobs.clear();
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public JobHolder findJobById(long id) {
+ return jobs.findById(id);
+ }
+
public final Comparator jobComparator = new Comparator() {
@Override
public int compare(JobHolder holder1, JobHolder holder2) {
diff --git a/jobqueue/src/com/path/android/jobqueue/persistentQueue/sqlite/SqlHelper.java b/jobqueue/src/com/path/android/jobqueue/persistentQueue/sqlite/SqlHelper.java
index ad5d689..50e3ad7 100644
--- a/jobqueue/src/com/path/android/jobqueue/persistentQueue/sqlite/SqlHelper.java
+++ b/jobqueue/src/com/path/android/jobqueue/persistentQueue/sqlite/SqlHelper.java
@@ -8,6 +8,9 @@
* Helper class for {@link SqliteJobQueue} to generate sql queries and statements.
*/
public class SqlHelper {
+
+ /**package**/ String FIND_BY_ID_QUERY;
+
private SQLiteStatement insertStatement;
private SQLiteStatement insertOrReplaceStatement;
private SQLiteStatement deleteStatement;
@@ -16,6 +19,7 @@ public class SqlHelper {
private SQLiteStatement nextJobDelayedUntilWithNetworkStatement;
private SQLiteStatement nextJobDelayedUntilWithoutNetworkStatement;
+
final SQLiteDatabase db;
final String tableName;
final String primaryKeyColumnName;
@@ -28,6 +32,7 @@ public SqlHelper(SQLiteDatabase db, String tableName, String primaryKeyColumnNam
this.columnCount = columnCount;
this.primaryKeyColumnName = primaryKeyColumnName;
this.sessionId = sessionId;
+ FIND_BY_ID_QUERY = "SELECT * FROM " + tableName + " WHERE " + DbOpenHelper.ID_COLUMN.columnName + " = ?";
}
public static String create(String tableName, Property primaryKey, Property... properties) {
diff --git a/jobqueue/src/com/path/android/jobqueue/persistentQueue/sqlite/SqliteJobQueue.java b/jobqueue/src/com/path/android/jobqueue/persistentQueue/sqlite/SqliteJobQueue.java
index 5f290e5..eefc667 100644
--- a/jobqueue/src/com/path/android/jobqueue/persistentQueue/sqlite/SqliteJobQueue.java
+++ b/jobqueue/src/com/path/android/jobqueue/persistentQueue/sqlite/SqliteJobQueue.java
@@ -159,6 +159,25 @@ public int countReadyJobs(boolean hasNetwork, Collection excludeGroups)
}
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public JobHolder findJobById(long id) {
+ Cursor cursor = db.rawQuery(sqlHelper.FIND_BY_ID_QUERY, new String[]{Long.toString(id)});
+ try {
+ if(!cursor.moveToFirst()) {
+ return null;
+ }
+ return createJobHolderFromCursor(cursor);
+ } catch (InvalidBaseJobException e) {
+ JqLog.e(e, "invalid job on findJobById");
+ return null;
+ } finally {
+ cursor.close();
+ }
+ }
+
/**
* {@inheritDoc}
*/
@@ -246,6 +265,9 @@ public Long getNextJobDelayUntilNs(boolean hasNetwork) {
}
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public void clear() {
sqlHelper.truncate();
diff --git a/jobqueue/test/com/path/android/jobqueue/test/jobmanager/AddInBackgroundTest.java b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/AddInBackgroundTest.java
index 20be3a3..454d3c5 100644
--- a/jobqueue/test/com/path/android/jobqueue/test/jobmanager/AddInBackgroundTest.java
+++ b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/AddInBackgroundTest.java
@@ -1,8 +1,14 @@
package com.path.android.jobqueue.test.jobmanager;
+import com.path.android.jobqueue.AsyncAddCallback;
+import com.path.android.jobqueue.BaseJob;
import com.path.android.jobqueue.Job;
+import com.path.android.jobqueue.JobHolder;
+import com.path.android.jobqueue.JobManager;
+import com.path.android.jobqueue.JobQueue;
import com.path.android.jobqueue.Params;
import com.path.android.jobqueue.test.jobs.DummyJob;
+import org.fest.reflect.core.*;
import org.hamcrest.*;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -14,15 +20,19 @@
@RunWith(RobolectricTestRunner.class)
public class AddInBackgroundTest extends JobManagerTestBase {
@Test
- public void testAddInBackground() {
- addInBackground(false);
- addInBackground(true);
-
+ public void testAddInBackground() throws InterruptedException {
+ for(boolean delay : new boolean[]{true, false}) {
+ for(boolean useCallback : new boolean[]{true, false}) {
+ addInBackground(delay, useCallback);
+ }
+ }
}
- public void addInBackground(boolean delayed) {
+
+ public void addInBackground(boolean delayed, boolean useCallback) throws InterruptedException {
long currentThreadId = Thread.currentThread().getId();
final AtomicLong onAddedThreadId = new AtomicLong();
final CountDownLatch addedLatch = new CountDownLatch(2);
+
Job dummyJob = new DummyJob(new Params(1).setDelayMs(delayed ? 1000 : 0)) {
@Override
public void onAdded() {
@@ -31,9 +41,33 @@ public void onAdded() {
addedLatch.countDown();
}
};
- createJobManager().addJobInBackground(dummyJob);
-
- addedLatch.countDown();
+ JobManager jobManager = createJobManager();
+ jobManager.stop();
+ final AtomicLong jobId = new AtomicLong(0);
+ if(useCallback) {
+ jobManager.addJobInBackground(dummyJob, new AsyncAddCallback() {
+ @Override
+ public void onAdded(long id) {
+ jobId.set(id);
+ addedLatch.countDown();
+ }
+ });
+ } else {
+ addedLatch.countDown();
+ jobManager.addJobInBackground(dummyJob);
+ }
+ addedLatch.await();
MatcherAssert.assertThat("thread ids should be different. delayed:" + delayed, currentThreadId, CoreMatchers.not(onAddedThreadId.get()));
+ if(useCallback) {
+ JobQueue queue = getNonPersistentQueue(jobManager);
+ JobHolder holder = queue.findJobById(jobId.longValue());
+ MatcherAssert.assertThat("there should be a job in the holder. id:" + jobId.longValue() +", delayed:" + delayed + ", use cb:" + useCallback
+ , holder, CoreMatchers.notNullValue());
+ MatcherAssert.assertThat("id callback should have the proper id:", holder.getBaseJob(), CoreMatchers.is((BaseJob) dummyJob));
+ }
+ }
+
+ protected JobQueue getNonPersistentQueue(JobManager jobManager) {
+ return Reflection.field("nonPersistentJobQueue").ofType(JobQueue.class).in(jobManager).get();
}
}
diff --git a/jobqueue/test/com/path/android/jobqueue/test/jobmanager/GroupingTest.java b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/GroupingTest.java
index 00103e9..0739d3c 100644
--- a/jobqueue/test/com/path/android/jobqueue/test/jobmanager/GroupingTest.java
+++ b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/GroupingTest.java
@@ -3,6 +3,7 @@
import com.path.android.jobqueue.JobHolder;
import com.path.android.jobqueue.JobManager;
import com.path.android.jobqueue.Params;
+import com.path.android.jobqueue.config.Configuration;
import com.path.android.jobqueue.test.jobs.DummyJob;
import org.fest.reflect.method.*;
import static org.hamcrest.CoreMatchers.*;
@@ -11,6 +12,9 @@
import org.junit.runner.RunWith;
import org.robolectric.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
@RunWith(RobolectricTestRunner.class)
public class GroupingTest extends JobManagerTestBase {
@Test
@@ -35,4 +39,68 @@ public void testGrouping() throws Exception {
removeJobMethod.invoke(group2Job);
MatcherAssert.assertThat("even after group2 job is complete, no jobs should be returned since we only have group1 jobs left", nextJobMethod.invoke(), is(nullValue()));
}
+
+ @Test
+ public void testGroupingRaceCondition() throws Exception {
+ DummyNetworkUtilWithConnectivityEventSupport dummyNetworkUtil = new DummyNetworkUtilWithConnectivityEventSupport();
+ JobManager jobManager = createJobManager(new Configuration.Builder(Robolectric.application)
+ .minConsumerCount(5).maxConsumerCount(10)
+ .networkUtil(dummyNetworkUtil));
+ dummyNetworkUtil.setHasNetwork(false, true);
+ //add a bunch of network requring jobs
+ final String GROUP_ID = "shared_group_id";
+ final int AFTER_ADDED_JOBS_COUNT = 5;
+ final int NOT_SET_JOB_ID = -1;
+ final AtomicInteger firstRunJob = new AtomicInteger(NOT_SET_JOB_ID);
+ final int FIRST_JOB_ID = -10;
+ final CountDownLatch onAddedCalled = new CountDownLatch(1);
+ final CountDownLatch remainingJobsOnAddedCalled = new CountDownLatch(AFTER_ADDED_JOBS_COUNT);
+ jobManager.addJobInBackground(new DummyJob(new Params(10).requireNetwork().groupBy(GROUP_ID)) {
+ @Override
+ public void onAdded() {
+ super.onAdded();
+ onAddedCalled.countDown();
+ try {
+ //wait until all other jobs are added
+ remainingJobsOnAddedCalled.await();
+ //wait a bit after all are added,
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ }
+ }
+
+ @Override
+ public void onRun() throws Throwable {
+ super.onRun();
+ firstRunJob.compareAndSet(NOT_SET_JOB_ID, FIRST_JOB_ID);
+ }
+ });
+ //ensure first jobs on added is called
+ onAddedCalled.await();
+ for(int i = 0; i < AFTER_ADDED_JOBS_COUNT; i ++) {
+ final int finalI = i;
+ jobManager.addJob(new DummyJob(new Params(5).groupBy(GROUP_ID).requireNetwork()) {
+ final int id = finalI + 1;
+
+ @Override
+ public void onAdded() {
+ super.onAdded();
+ remainingJobsOnAddedCalled.countDown();
+ }
+
+ @Override
+ public void onRun() throws Throwable {
+ super.onRun();
+ firstRunJob.compareAndSet(NOT_SET_JOB_ID, id);
+ }
+ });
+ }
+ dummyNetworkUtil.setHasNetwork(true, true);
+ //wait until all jobs are completed
+ while(firstRunJob.get() == NOT_SET_JOB_ID) {
+ Thread.sleep(100);
+ }
+ MatcherAssert.assertThat("highest priority job should run if it is added before others", firstRunJob.get(), is(FIRST_JOB_ID));
+
+ }
}
diff --git a/jobqueue/test/com/path/android/jobqueue/test/jobmanager/InjectorTest.java b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/InjectorTest.java
index ac6152f..7129710 100644
--- a/jobqueue/test/com/path/android/jobqueue/test/jobmanager/InjectorTest.java
+++ b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/InjectorTest.java
@@ -6,6 +6,7 @@
import com.path.android.jobqueue.Params;
import com.path.android.jobqueue.config.Configuration;
import com.path.android.jobqueue.di.DependencyInjector;
+import com.path.android.jobqueue.log.CustomLogger;
import com.path.android.jobqueue.test.jobs.DummyJob;
import static org.hamcrest.CoreMatchers.*;
import org.hamcrest.*;
@@ -13,6 +14,8 @@
import org.junit.runner.RunWith;
import org.robolectric.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@RunWith(RobolectricTestRunner.class)
@@ -42,6 +45,55 @@ public void inject(BaseJob job) {
holder = getNextJobMethod(jobManager).invoke();
MatcherAssert.assertThat("injection should be called for persistent job", holder.getBaseJob(), equalTo(injectedJobReference.getObject()));
MatcherAssert.assertThat("injection should be called two times for persistent job", injectionCallCount.get(), equalTo(3));
+ }
+
+ @Test
+ public void testInjectorCrash() throws Exception {
+ final String EXCEPTION_MESSAGE = "could not inject for whatever reason :)";
+ DependencyInjector dummyDependencyInjector = new DependencyInjector() {
+ @Override
+ public void inject(BaseJob baseJob) {
+ throw new RuntimeException(EXCEPTION_MESSAGE);
+ }
+ };
+
+ final ObjectReference objectReference = new ObjectReference();
+ final CountDownLatch exceptionLatch = new CountDownLatch(1);
+ CustomLogger customLogger = new CustomLogger() {
+ @Override
+ public boolean isDebugEnabled() {
+ return false;
+ }
+
+ @Override
+ public void d(String s, Object... objects) {
+
+ }
+
+ @Override
+ public void e(Throwable throwable, String s, Object... objects) {
+ objectReference.setObject(throwable);
+ exceptionLatch.countDown();
+ }
+
+ @Override
+ public void e(String s, Object... objects) {
+ //
+ }
+ };
+ JobManager jobManager = createJobManager(new Configuration.Builder(Robolectric.application).injector(dummyDependencyInjector).customLogger(customLogger));
+ Throwable addException = null;
+ try {
+ jobManager.addJob(new DummyJob(new Params(0)));
+ } catch (Throwable t) {
+ addException = t;
+ }
+ MatcherAssert.assertThat("addJob should throw exception if dependency injector throws exception", addException, notNullValue());
+ jobManager.addJobInBackground(new DummyJob(new Params(0)));
+ exceptionLatch.await(2, TimeUnit.SECONDS);
+ MatcherAssert.assertThat("there should be a received exception", objectReference.getObject(), notNullValue());
+ MatcherAssert.assertThat("logged exception should be a runtime exception", objectReference.getObject(), instanceOf(RuntimeException.class));
+ MatcherAssert.assertThat("logged exception should have expected message", ((Throwable)objectReference.getObject()).getMessage(), is(EXCEPTION_MESSAGE));
}
}
diff --git a/jobqueue/test/com/path/android/jobqueue/test/jobmanager/JobManagerTestBase.java b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/JobManagerTestBase.java
index b44fc39..d416b76 100644
--- a/jobqueue/test/com/path/android/jobqueue/test/jobmanager/JobManagerTestBase.java
+++ b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/JobManagerTestBase.java
@@ -32,6 +32,31 @@ protected JobManager createJobManager(Configuration.Builder configurationBuilder
+ protected static class DummyTwoLatchJob extends DummyJob {
+ private final CountDownLatch waitFor;
+ private final CountDownLatch trigger;
+ private final CountDownLatch onRunLatch;
+
+ protected DummyTwoLatchJob(Params params, CountDownLatch waitFor, CountDownLatch trigger) {
+ super(params);
+ this.waitFor = waitFor;
+ this.trigger = trigger;
+ onRunLatch = new CountDownLatch(1);
+ }
+
+ public void waitTillOnRun() throws InterruptedException {
+ onRunLatch.await();
+ }
+
+ @Override
+ public void onRun() throws Throwable {
+ onRunLatch.countDown();
+ waitFor.await();
+ super.onRun();
+ trigger.countDown();
+ }
+ }
+
protected static class DummyLatchJob extends DummyJob {
private final CountDownLatch latch;
@@ -108,6 +133,10 @@ public boolean isConnected(Context context) {
public void setListener(Listener listener) {
this.listener = listener;
}
+
+ public boolean isConnected() {
+ return hasNetwork;
+ }
}
protected static class ObjectReference {
diff --git a/jobqueue/test/com/path/android/jobqueue/test/jobmanager/JobStatusTest.java b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/JobStatusTest.java
new file mode 100644
index 0000000..28656ba
--- /dev/null
+++ b/jobqueue/test/com/path/android/jobqueue/test/jobmanager/JobStatusTest.java
@@ -0,0 +1,119 @@
+package com.path.android.jobqueue.test.jobmanager;
+
+import com.path.android.jobqueue.Job;
+import com.path.android.jobqueue.JobManager;
+import com.path.android.jobqueue.JobStatus;
+import com.path.android.jobqueue.Params;
+import com.path.android.jobqueue.config.Configuration;
+import com.path.android.jobqueue.test.jobs.DummyJob;
+import static org.hamcrest.CoreMatchers.*;
+import static org.hamcrest.MatcherAssert.*;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.robolectric.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
+@RunWith(RobolectricTestRunner.class)
+public class JobStatusTest extends JobManagerTestBase {
+ @Test
+ public void testJobStatus() throws InterruptedException {
+ DummyNetworkUtilWithConnectivityEventSupport networkUtil = new DummyNetworkUtilWithConnectivityEventSupport();
+ networkUtil.setHasNetwork(false, true);
+ JobManager jobManager = createJobManager(new Configuration.Builder(Robolectric.application).networkUtil(networkUtil));
+ jobManager.stop();
+ List networkRequiringJobIndices = new ArrayList();
+ Job[] jobs = new Job[] {
+ new DummyJob(new Params(0)),
+ new DummyJob(new Params(0).persist()),
+ new DummyJob(new Params(0).persist().requireNetwork())
+ };
+ long[] ids = new long[jobs.length];
+ for(int i = 0; i < jobs.length; i ++) {
+ ids[i] = jobManager.addJob(jobs[i]);
+ if(jobs[i].requiresNetwork()) {
+ networkRequiringJobIndices.add(i);
+ }
+ JobStatus expectedStatus = (networkUtil.isConnected() || jobs[i].requiresNetwork() == false) ? JobStatus.WAITING_READY :
+ JobStatus.WAITING_NOT_READY;
+ assertThat("job should have correct status after being added",
+ jobManager.getJobStatus(ids[i], jobs[i].isPersistent()), is(expectedStatus));
+ }
+
+ //create an unknown id, ensure status for that
+
+ boolean exists;
+ long unknownId;
+ do {
+ unknownId = (long) (Math.random() * 10000 - 5000);
+ exists = false;
+ for(long id : ids) {
+ if(id == unknownId) {
+ exists = true;
+ continue;
+ }
+ }
+ } while (exists);
+ for(boolean persistent : new boolean[]{true, false}) {
+ assertThat("job with unknown id should return as expected", jobManager.getJobStatus(unknownId, persistent), is(JobStatus.UNKNOWN));
+ }
+
+ CountDownLatch startLatch = new CountDownLatch(1), endLatch = new CountDownLatch(1);
+ DummyTwoLatchJob twoLatchJob = new DummyTwoLatchJob(new Params(0), startLatch, endLatch);
+ jobManager.start();
+ long jobId = jobManager.addJob(twoLatchJob);
+ twoLatchJob.waitTillOnRun();
+ assertThat("job should be in running state", jobManager.getJobStatus(jobId, false), is(JobStatus.RUNNING));
+ startLatch.countDown();//let it run
+ endLatch.await();//wait till it finishes
+ Thread.sleep(500);//give some time to job manager to clear the job
+ assertThat("finished job should go to unknown state", jobManager.getJobStatus(jobId, false), is(JobStatus.UNKNOWN));
+
+ //network requiring job should not be ready
+ for(Integer i : networkRequiringJobIndices) {
+ assertThat("network requiring job should still be not-ready", jobManager.getJobStatus(ids[i], jobs[i].isPersistent()), is(JobStatus.WAITING_NOT_READY));
+ }
+ jobManager.stop();
+ networkUtil.setHasNetwork(true, true);
+ for(Integer i : networkRequiringJobIndices) {
+ assertThat("network requiring job should still be ready after network is there", jobManager.getJobStatus(ids[i], jobs[i].isPersistent()), is(JobStatus.WAITING_READY));
+ }
+
+ jobManager.start();
+ int limit = 10;
+ while (jobManager.count() > 0 && limit-- > 0) {
+ Thread.sleep(1000);
+ }
+ assertThat("jobs should finish", jobManager.count(), is(0));
+ for(int i = 0; i < jobs.length; i ++) {
+ //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));
+ }
+ }
+ }
+}
diff --git a/jobqueue/test/com/path/android/jobqueue/test/jobqueue/JobQueueTestBase.java b/jobqueue/test/com/path/android/jobqueue/test/jobqueue/JobQueueTestBase.java
index 9c3c188..2805e54 100644
--- a/jobqueue/test/com/path/android/jobqueue/test/jobqueue/JobQueueTestBase.java
+++ b/jobqueue/test/com/path/android/jobqueue/test/jobqueue/JobQueueTestBase.java
@@ -9,7 +9,7 @@
import com.path.android.jobqueue.test.util.JobQueueFactory;
import org.fest.reflect.core.*;
import static org.hamcrest.CoreMatchers.*;
-import org.hamcrest.*;
+import static org.hamcrest.MatcherAssert.*;
import org.junit.Ignore;
import org.junit.Test;
@@ -28,34 +28,34 @@ public JobQueueTestBase(JobQueueFactory factory) {
public void testBasicAddRemoveCount() throws Exception {
final int ADD_COUNT = 6;
JobQueue jobQueue = createNewJobQueue();
- MatcherAssert.assertThat((int) jobQueue.count(), equalTo(0));
- MatcherAssert.assertThat(jobQueue.nextJobAndIncRunCount(true, null), nullValue());
+ assertThat((int) jobQueue.count(), equalTo(0));
+ assertThat(jobQueue.nextJobAndIncRunCount(true, null), nullValue());
for (int i = 0; i < ADD_COUNT; i++) {
JobHolder holder = createNewJobHolder();
jobQueue.insert(holder);
- MatcherAssert.assertThat((int) jobQueue.count(), equalTo(i + 1));
- MatcherAssert.assertThat(holder.getId(), notNullValue());
+ assertThat((int) jobQueue.count(), equalTo(i + 1));
+ assertThat(holder.getId(), notNullValue());
jobQueue.insertOrReplace(holder);
- MatcherAssert.assertThat((int) jobQueue.count(), equalTo(i + 1));
+ assertThat((int) jobQueue.count(), equalTo(i + 1));
}
JobHolder firstHolder = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat(firstHolder.getRunCount(), equalTo(1));
+ assertThat(firstHolder.getRunCount(), equalTo(1));
//size should be down 1
- MatcherAssert.assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 1));
+ assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 1));
//should return another job
JobHolder secondHolder = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat(secondHolder.getRunCount(), equalTo(1));
+ assertThat(secondHolder.getRunCount(), equalTo(1));
//size should be down 2
- MatcherAssert.assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 2));
+ assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 2));
//second holder and first holder should have different ids
- MatcherAssert.assertThat(firstHolder.getId(), not(secondHolder.getId()));
+ assertThat(firstHolder.getId(), not(secondHolder.getId()));
jobQueue.remove(secondHolder);
- MatcherAssert.assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 2));
+ assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 2));
jobQueue.remove(secondHolder);
//non existed job removed, count should be the same
- MatcherAssert.assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 2));
+ assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 2));
jobQueue.remove(firstHolder);
- MatcherAssert.assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 2));
+ assertThat((int) jobQueue.count(), equalTo(ADD_COUNT - 2));
}
@Test
@@ -70,11 +70,12 @@ public void testPriority() throws Exception {
int minPriority = Integer.MAX_VALUE;
for (int i = 0; i < JOB_LIMIT; i++) {
JobHolder holder = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat(holder.getPriority() <= minPriority, is(true));
+ assertThat(holder.getPriority() <= minPriority, is(true));
}
- MatcherAssert.assertThat(jobQueue.nextJobAndIncRunCount(true, null), nullValue());
+ assertThat(jobQueue.nextJobAndIncRunCount(true, null), nullValue());
}
+
@Test
public void testDelayUntilWithPriority() throws Exception {
JobQueue jobQueue = createNewJobQueue();
@@ -83,7 +84,7 @@ public void testDelayUntilWithPriority() throws Exception {
JobHolder highPriorityHolder = createNewJobHolderWithDelayUntil(new Params(10), now + 20000 * JobManager.NS_PER_MS);
jobQueue.insert(lowPriorityHolder);
jobQueue.insert(highPriorityHolder);
- MatcherAssert.assertThat("when asked, if lower priority job has less delay until, we should return it",
+ assertThat("when asked, if lower priority job has less delay until, we should return it",
jobQueue.getNextJobDelayUntilNs(true), equalTo(lowPriorityHolder.getDelayUntilNs()));
}
@@ -97,20 +98,20 @@ public void testGroupId() throws Exception {
long jobId4 = jobQueue.insert(createNewJobHolder(new Params(0).groupBy("group2")));
long jobId5 = jobQueue.insert(createNewJobHolder(new Params(0).groupBy("group1")));
JobHolder holder1 = jobQueue.nextJobAndIncRunCount(true, Arrays.asList(new String[]{"group2"}));
- MatcherAssert.assertThat("first jobs should be from group group2 if group1 is excluded",
+ assertThat("first jobs should be from group group2 if group1 is excluded",
holder1.getBaseJob().getRunGroupId(), equalTo("group1"));
- MatcherAssert.assertThat("correct job should be returned if groupId is provided",
+ assertThat("correct job should be returned if groupId is provided",
holder1.getId(), equalTo(jobId1));
- MatcherAssert.assertThat("no jobs should be returned if all groups are excluded",
+ assertThat("no jobs should be returned if all groups are excluded",
jobQueue.nextJobAndIncRunCount(true,
Arrays.asList(new String[]{"group1", "group2"})),
is(nullValue()));
long jobId6 = jobQueue.insert(createNewJobHolder(new Params(0)));
- MatcherAssert.assertThat("both groups are disabled, null group job should be returned",
+ assertThat("both groups are disabled, null group job should be returned",
jobQueue.nextJobAndIncRunCount(true,
Arrays.asList(new String[]{"group1", "group2"})).getId(),
is(jobId6));
- MatcherAssert.assertThat("if group1 is excluded, next job should be from group2",
+ assertThat("if group1 is excluded, next job should be from group2",
jobQueue.nextJobAndIncRunCount(true, Arrays.asList(new String[]{"group1"})).getBaseJob().getRunGroupId()
, equalTo("group2"));
@@ -118,12 +119,12 @@ public void testGroupId() throws Exception {
jobQueue.insertOrReplace(holder1);
//ask for it again, should return the same holder because it is grouped
JobHolder holder2 = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat("for grouped jobs, re-fetching job should work fine",
+ assertThat("for grouped jobs, re-fetching job should work fine",
holder2.getId(), equalTo(holder1.getId()));
JobHolder holder3 = jobQueue.nextJobAndIncRunCount(true,
Arrays.asList(new String[]{"group1"}));
- MatcherAssert.assertThat("if a group it excluded, next available from another group should be returned",
+ assertThat("if a group it excluded, next available from another group should be returned",
holder3.getId(), equalTo(jobId4));
//add two more non-grouped jobs
@@ -131,12 +132,12 @@ public void testGroupId() throws Exception {
long jobId8 = jobQueue.insert(createNewJobHolder(new Params(0)));
JobHolder holder4 = jobQueue.nextJobAndIncRunCount(true,
Arrays.asList(new String[]{"group1", "group2"}));
- MatcherAssert.assertThat("if all grouped jobs are excluded, non-grouped jobs should be returned",
+ assertThat("if all grouped jobs are excluded, non-grouped jobs should be returned",
holder4.getId(),
equalTo(jobId7));
jobQueue.insertOrReplace(holder4);
//for non-grouped jobs, run counts should be respected
- MatcherAssert.assertThat("if all grouped jobs are excluded, re-inserted highest priority job should still be returned",
+ assertThat("if all grouped jobs are excluded, re-inserted highest priority job should still be returned",
jobQueue.nextJobAndIncRunCount(true,
Arrays.asList(new String[]{"group1", "group2"})).getId(),
equalTo(jobId7));
@@ -153,13 +154,13 @@ public void testDueDelayUntilWithPriority() throws Exception {
long soonJobDelay = 2000;
JobHolder highestPriorityDelayedJob = createNewJobHolderWithDelayUntil(new Params(12), now + soonJobDelay * JobManager.NS_PER_MS);
long highestPriorityDelayedJobId = jobQueue.insert(highestPriorityDelayedJob);
- MatcherAssert.assertThat("when asked, if job's due has passed, highest priority jobs's delay until should be " +
+ assertThat("when asked, if job's due has passed, highest priority jobs's delay until should be " +
"returned",
jobQueue.getNextJobDelayUntilNs(true), equalTo(highPriorityHolder.getDelayUntilNs()));
//make sure soon job is valid now
Thread.sleep(soonJobDelay);
- MatcherAssert.assertThat("when a job's time come, it should be returned",
+ assertThat("when a job's time come, it should be returned",
jobQueue.nextJobAndIncRunCount(true, null).getId(), equalTo(highestPriorityDelayedJobId));
}
@@ -174,16 +175,16 @@ public void testDelayUntil() throws Exception {
jobQueue.insert(networkJobHolder);
jobQueue.insert(noNetworkJobHolder);
- MatcherAssert.assertThat("if there is no network, delay until should be provided for no network job",
+ assertThat("if there is no network, delay until should be provided for no network job",
jobQueue.getNextJobDelayUntilNs(false), equalTo(noNetworkJobHolder.getDelayUntilNs()));
- MatcherAssert.assertThat("if there is network, delay until should be provided for network job because it is " +
+ assertThat("if there is network, delay until should be provided for network job because it is " +
"sooner", jobQueue.getNextJobDelayUntilNs(true), equalTo(networkJobHolder.getDelayUntilNs()));
JobHolder noNetworkJobHolder2 = createNewJobHolderWithDelayUntil(new Params(0), now + 100000 * JobManager.NS_PER_MS);
jobQueue.insert(noNetworkJobHolder2);
- MatcherAssert.assertThat("if there is network, any job's delay until should be returned",
+ assertThat("if there is network, any job's delay until should be returned",
jobQueue.getNextJobDelayUntilNs(true), equalTo(noNetworkJobHolder2.getDelayUntilNs()));
}
@@ -194,13 +195,13 @@ public void testTruncate() throws Exception {
for(int i = 0; i < LIMIT; i ++) {
jobQueue.insert(createNewJobHolder());
}
- MatcherAssert.assertThat("queue should have all jobs", jobQueue.count(), equalTo(LIMIT));
+ assertThat("queue should have all jobs", jobQueue.count(), equalTo(LIMIT));
jobQueue.clear();
- MatcherAssert.assertThat("after clear, queue should be empty", jobQueue.count(), equalTo(0));
+ assertThat("after clear, queue should be empty", jobQueue.count(), equalTo(0));
for(int i = 0; i < LIMIT; i ++) {
jobQueue.insert(createNewJobHolder());
}
- MatcherAssert.assertThat("if we add jobs again, count should match", jobQueue.count(), equalTo(LIMIT));
+ assertThat("if we add jobs again, count should match", jobQueue.count(), equalTo(LIMIT));
}
@Test
@@ -230,8 +231,8 @@ public void testPriorityWithDelayedJobs() throws Exception {
int lastPriority = Integer.MAX_VALUE;
for(int i = 0; i < 5; i++) {
JobHolder next = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat("next job should not be null", next, notNullValue());
- MatcherAssert.assertThat("next job's priority should be lower then previous for job " + i, next.getPriority() <= lastPriority, is(true));
+ assertThat("next job should not be null", next, notNullValue());
+ assertThat("next job's priority should be lower then previous for job " + i, next.getPriority() <= lastPriority, is(true));
lastPriority = next.getPriority();
}
@@ -260,7 +261,7 @@ public void testSessionId() throws Exception {
JobHolder jobHolder = createNewJobHolder();
jobQueue.insert(jobHolder);
jobHolder = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat("session id should be attached to next job",
+ assertThat("session id should be attached to next job",
jobHolder.getRunningSessionId(), equalTo(sessionId));
}
@@ -276,10 +277,10 @@ public void testPriorityWithReAdd() throws Exception {
int minPriority = Integer.MAX_VALUE;
for (int i = 0; i < JOB_LIMIT; i++) {
JobHolder holder = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat(holder.getPriority() <= minPriority, is(true));
+ assertThat(holder.getPriority() <= minPriority, is(true));
jobQueue.insertOrReplace(holder);
}
- MatcherAssert.assertThat(jobQueue.nextJobAndIncRunCount(true, null), notNullValue());
+ assertThat(jobQueue.nextJobAndIncRunCount(true, null), notNullValue());
}
@Test
@@ -288,10 +289,10 @@ public void testRemove() throws Exception {
JobHolder holder = createNewJobHolder();
jobQueue.insert(holder);
Long jobId = holder.getId();
- MatcherAssert.assertThat(jobQueue.nextJobAndIncRunCount(true, null).getId(), equalTo(jobId));
- MatcherAssert.assertThat(jobQueue.nextJobAndIncRunCount(true, null), is(nullValue()));
+ assertThat(jobQueue.nextJobAndIncRunCount(true, null).getId(), equalTo(jobId));
+ assertThat(jobQueue.nextJobAndIncRunCount(true, null), is(nullValue()));
jobQueue.remove(holder);
- MatcherAssert.assertThat(jobQueue.nextJobAndIncRunCount(true, null), is(nullValue()));
+ assertThat(jobQueue.nextJobAndIncRunCount(true, null), is(nullValue()));
}
@Test
@@ -299,15 +300,15 @@ public void testNetwork() throws Exception {
JobQueue jobQueue = createNewJobQueue();
JobHolder jobHolder = createNewJobHolder(new Params(0));
jobQueue.insert(jobHolder);
- MatcherAssert.assertThat("no network job should be returned even if there is no netowrk",
+ assertThat("no network job should be returned even if there is no netowrk",
jobQueue.nextJobAndIncRunCount(false, null), notNullValue());
jobQueue.remove(jobHolder);
jobHolder = createNewJobHolder(new Params(0).requireNetwork());
- MatcherAssert.assertThat("if there isn't any network, job with network requirement should not return",
+ assertThat("if there isn't any network, job with network requirement should not return",
jobQueue.nextJobAndIncRunCount(false, null), nullValue());
- MatcherAssert.assertThat("if there is network, job with network requirement should be returned",
+ assertThat("if there is network, job with network requirement should be returned",
jobQueue.nextJobAndIncRunCount(true, null), nullValue());
jobQueue.remove(jobHolder);
@@ -317,17 +318,17 @@ public void testNetwork() throws Exception {
long firstJobId = jobQueue.insert(jobHolder);
long secondJobId = jobQueue.insert(jobHolder2);
JobHolder retrieved = jobQueue.nextJobAndIncRunCount(false, null);
- MatcherAssert.assertThat("one job should be returned w/o network", retrieved, notNullValue());
+ assertThat("one job should be returned w/o network", retrieved, notNullValue());
if(retrieved != null) {
- MatcherAssert.assertThat("no network job should be returned although it has lower priority", retrieved.getId(), equalTo(firstJobId));
+ assertThat("no network job should be returned although it has lower priority", retrieved.getId(), equalTo(firstJobId));
}
- MatcherAssert.assertThat("no other job should be returned w/o network", jobQueue.nextJobAndIncRunCount(false, null), nullValue());
+ assertThat("no other job should be returned w/o network", jobQueue.nextJobAndIncRunCount(false, null), nullValue());
retrieved = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat("if network is back, network requiring job should be returned", retrieved, notNullValue());
+ assertThat("if network is back, network requiring job should be returned", retrieved, notNullValue());
if(retrieved != null) {
- MatcherAssert.assertThat("when there is network, network job should be returned", retrieved.getId(), equalTo(secondJobId));
+ assertThat("when there is network, network job should be returned", retrieved.getId(), equalTo(secondJobId));
}
//add first job back
jobQueue.insertOrReplace(jobHolder);
@@ -335,18 +336,18 @@ public void testNetwork() throws Exception {
jobQueue.insertOrReplace(jobHolder2);
retrieved = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat("if network is back, job w/ higher priority should be returned", retrieved, notNullValue());
+ assertThat("if network is back, job w/ higher priority should be returned", retrieved, notNullValue());
if(retrieved != null) {
- MatcherAssert.assertThat("if network is back, job w/ higher priority should be returned", retrieved.getId(), equalTo(secondJobId));
+ assertThat("if network is back, job w/ higher priority should be returned", retrieved.getId(), equalTo(secondJobId));
}
jobQueue.insertOrReplace(jobHolder2);
JobHolder highestPriorityJob = createNewJobHolder(new Params(10));
long highestPriorityJobId = jobQueue.insert(highestPriorityJob);
retrieved = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat("w/ or w/o network, highest priority should be returned", retrieved, notNullValue());
+ assertThat("w/ or w/o network, highest priority should be returned", retrieved, notNullValue());
if(retrieved != null) {
- MatcherAssert.assertThat("w/ or w/o network, highest priority should be returned", retrieved.getId(), equalTo(highestPriorityJobId));
+ assertThat("w/ or w/o network, highest priority should be returned", retrieved.getId(), equalTo(highestPriorityJobId));
}
//TODO test delay until
@@ -355,40 +356,40 @@ public void testNetwork() throws Exception {
@Test
public void testCountReadyJobs() throws Exception {
JobQueue jobQueue = createNewJobQueue();
- MatcherAssert.assertThat("initial count should be 0 for ready jobs", jobQueue.countReadyJobs(true, null), equalTo(0));
+ assertThat("initial count should be 0 for ready jobs", jobQueue.countReadyJobs(true, null), equalTo(0));
//add some jobs
jobQueue.insert(createNewJobHolder());
jobQueue.insert(createNewJobHolder(new Params(0).requireNetwork()));
long now = System.nanoTime();
long delay = 1000;
jobQueue.insert(createNewJobHolderWithDelayUntil(new Params(0), now + TimeUnit.MILLISECONDS.toNanos(delay)));
- MatcherAssert.assertThat("ready count should be 1 if there is no network", jobQueue.countReadyJobs(false, null), equalTo(1));
- MatcherAssert.assertThat("ready count should be 2 if there is network", jobQueue.countReadyJobs(true, null), equalTo(2));
+ assertThat("ready count should be 1 if there is no network", jobQueue.countReadyJobs(false, null), equalTo(1));
+ assertThat("ready count should be 2 if there is network", jobQueue.countReadyJobs(true, null), equalTo(2));
Thread.sleep(delay);
- MatcherAssert.assertThat("when needed delay time passes, ready count should be 3", jobQueue.countReadyJobs(true, null), equalTo(3));
- MatcherAssert.assertThat("when needed delay time passes but no network, ready count should be 2", jobQueue.countReadyJobs(false, null), equalTo(2));
+ assertThat("when needed delay time passes, ready count should be 3", jobQueue.countReadyJobs(true, null), equalTo(3));
+ assertThat("when needed delay time passes but no network, ready count should be 2", jobQueue.countReadyJobs(false, null), equalTo(2));
jobQueue.insert(createNewJobHolder(new Params(5).groupBy("group1")));
jobQueue.insert(createNewJobHolder(new Params(5).groupBy("group1")));
- MatcherAssert.assertThat("when more than 1 job from same group is created, ready jobs should increment only by 1",
+ assertThat("when more than 1 job from same group is created, ready jobs should increment only by 1",
jobQueue.countReadyJobs(true, null), equalTo(4));
- MatcherAssert.assertThat("excluding groups should work",
+ assertThat("excluding groups should work",
jobQueue.countReadyJobs(true, Arrays.asList(new String[]{"group1"})), equalTo(3));
- MatcherAssert.assertThat("giving a non-existing group should not fool the count",
+ assertThat("giving a non-existing group should not fool the count",
jobQueue.countReadyJobs(true, Arrays.asList(new String[]{"group3423"})), equalTo(4));
jobQueue.insert(createNewJobHolder(new Params(3).groupBy("group2")));
- MatcherAssert.assertThat("when a job from another group is added, ready job count should inc",
+ assertThat("when a job from another group is added, ready job count should inc",
jobQueue.countReadyJobs(true, null), equalTo(5));
now = System.nanoTime();
jobQueue.insert(createNewJobHolderWithDelayUntil(new Params(3).groupBy("group3"), now + TimeUnit.MILLISECONDS.toNanos(delay)));
- MatcherAssert.assertThat("when a delayed job from another group is added, ready count should not change",
+ assertThat("when a delayed job from another group is added, ready count should not change",
jobQueue.countReadyJobs(true, null), equalTo(5));
jobQueue.insert(createNewJobHolder(new Params(3).groupBy("group3")));
- MatcherAssert.assertThat("when another job from delayed group is added, ready job count should inc",
+ assertThat("when another job from delayed group is added, ready job count should inc",
jobQueue.countReadyJobs(true, null), equalTo(6));
Thread.sleep(delay);
- MatcherAssert.assertThat("when delay passes and a job from existing group becomes available, ready job count should not change",
+ assertThat("when delay passes and a job from existing group becomes available, ready job count should not change",
jobQueue.countReadyJobs(true, null), equalTo(6));
- MatcherAssert.assertThat("when some groups are excluded, count should be correct",
+ assertThat("when some groups are excluded, count should be correct",
jobQueue.countReadyJobs(true, Arrays.asList(new String[]{"group1", "group3"})), equalTo(4));
//jobs w/ same group id but with different persistence constraints should not fool the count
@@ -397,11 +398,11 @@ public void testCountReadyJobs() throws Exception {
jobQueue.insert(createNewJobHolderWithDelayUntil(new Params(0).groupBy("group10"), now + 1000));
jobQueue.insert(createNewJobHolderWithDelayUntil(new Params(0).persist().groupBy("group10"), now - 1000));
jobQueue.insert(createNewJobHolderWithDelayUntil(new Params(0).groupBy("group10"), now - 1000));
- MatcherAssert.assertThat("when many jobs are added w/ different constraints but same group id, ready count should not be fooled",
+ assertThat("when many jobs are added w/ different constraints but same group id, ready count should not be fooled",
jobQueue.countReadyJobs(true, Arrays.asList(new String[]{"group1", "group3"})), equalTo(5));
- MatcherAssert.assertThat("when many jobs are added w/ different constraints but same group id, ready count should not be fooled",
+ assertThat("when many jobs are added w/ different constraints but same group id, ready count should not be fooled",
jobQueue.countReadyJobs(true, null), equalTo(7));
- MatcherAssert.assertThat("when many jobs are added w/ different constraints but same group id, ready count should not be fooled",
+ assertThat("when many jobs are added w/ different constraints but same group id, ready count should not be fooled",
jobQueue.countReadyJobs(false, Arrays.asList(new String[]{"group1", "group3"})), equalTo(4));
}
@@ -424,20 +425,72 @@ public void testJobFields() throws Exception {
for (int i = 0; i < 2; i++) {
JobHolder received = jobQueue.nextJobAndIncRunCount(true, null);
- MatcherAssert.assertThat("job id should be preserved", received.getId(), equalTo(id));
- MatcherAssert.assertThat("job priority should be preserved", received.getPriority(), equalTo(priority));
- MatcherAssert.assertThat("job session id should be assigned", received.getRunningSessionId(), equalTo(sessionId));
- MatcherAssert.assertThat("job run count should be incremented", received.getRunCount(), equalTo(runCount + i + 1));
+ assertThat("job id should be preserved", received.getId(), equalTo(id));
+ assertThat("job priority should be preserved", received.getPriority(), equalTo(priority));
+ assertThat("job session id should be assigned", received.getRunningSessionId(), equalTo(sessionId));
+ assertThat("job run count should be incremented", received.getRunCount(), equalTo(runCount + i + 1));
jobQueue.insertOrReplace(received);
}
}
+ private void assertJob(JobQueue jobQueue, String msg, long id, /*nullable*/ JobHolder holder) {
+ if(holder == null) {
+ assertThat(msg, jobQueue.findJobById(id), nullValue());
+ return;
+ }
+ assertThat(msg + "(existence check)", jobQueue.findJobById(id), notNullValue());
+ assertThat(msg + "(id check)", jobQueue.findJobById(id).getId(), is(holder.getId()));
+ }
+
+ @Test
+ public void testFindJobHolderById() {
+ JobQueue jobQueue = createNewJobQueue();
+ assertJob(jobQueue, "non existing job (negative id)", -4, null);
+ assertJob(jobQueue, "non existing job (positive id)", +4, null);
+ final int LIMIT = 100;
+ JobHolder[] holders = new JobHolder[LIMIT];
+ long[] ids = new long[LIMIT];
+ for(int i = 0; i < LIMIT; i++) {
+ holders[i] = createNewJobHolder(new Params((int) (Math.random() * 50)).setPersistent(Math.random() < .5).setRequiresNetwork(Math.random() < .5));
+ ids[i] = jobQueue.insert(holders[i]);
+ assertJob(jobQueue, "job by id should work for inserted job", ids[i], holders[i]);
+ }
+ final int REMOVE_CNT = LIMIT / 2;
+ for(int i = 0; i < REMOVE_CNT; i++) {
+ int ind = (int) (Math.random() * LIMIT);
+ if(holders[ind] == null) {
+ continue;
+ }
+ //remove some randomly, up to half
+ jobQueue.remove(holders[ind]);
+ holders[ind] = null;
+ }
+ //re-query all, ensure we can still find non-removed jobs and not find removed jobs
+ for(int i = 0; i < LIMIT; i++) {
+ if(holders[i] != null) {
+ assertJob(jobQueue, "if job is still in the Q, it should be returned", ids[i], holders[i]);
+ //re add job
+ jobQueue.insertOrReplace(holders[i]);
+ //re-test after re-add
+ assertJob(jobQueue, "after re-insert, if job is still in the Q, it should be returned", ids[i], holders[i]);
+ } else {
+ assertJob(jobQueue, "removed job should not be returned in id query", ids[i], null);
+ }
+ }
+ jobQueue.clear();
+ for(int i = 0; i < LIMIT; i++) {
+ assertJob(jobQueue, "after clear, find by id should return null", ids[i], null);
+ }
+ }
+
protected JobHolder createNewJobHolder() {
return createNewJobHolder(new Params(0));
}
protected JobHolder createNewJobHolder(Params params) {
- return new JobHolder(null, getPriorityField(params).get(), getGroupIdField(params).get(), 0, new DummyJob(params), System.nanoTime(), Long.MIN_VALUE, JobManager.NOT_RUNNING_SESSION_ID);
+ long delay = getDelayMsField(params).get();
+ return new JobHolder(null, getPriorityField(params).get(), getGroupIdField(params).get(), 0, new DummyJob(params), System.nanoTime(),
+ delay > 0 ? System.nanoTime() + delay * JobManager.NS_PER_MS : JobManager.NOT_DELAYED_JOB_DELAY, JobManager.NOT_RUNNING_SESSION_ID);
}
private JobHolder createNewJobHolderWithDelayUntil(Params params, long delayUntil) {
@@ -446,7 +499,7 @@ private JobHolder createNewJobHolderWithDelayUntil(Params params, long delayUnti
return jobHolder;
}
- private JobQueue createNewJobQueue() {
+ protected JobQueue createNewJobQueue() {
return createNewJobQueueWithSessionId(System.nanoTime());
}
diff --git a/jobqueue/test/com/path/android/jobqueue/test/jobqueue/NonPersistentJobQueueTest.java b/jobqueue/test/com/path/android/jobqueue/test/jobqueue/NonPersistentJobQueueTest.java
index 32ca824..aec3924 100644
--- a/jobqueue/test/com/path/android/jobqueue/test/jobqueue/NonPersistentJobQueueTest.java
+++ b/jobqueue/test/com/path/android/jobqueue/test/jobqueue/NonPersistentJobQueueTest.java
@@ -1,10 +1,16 @@
package com.path.android.jobqueue.test.jobqueue;
+import com.path.android.jobqueue.JobHolder;
+import com.path.android.jobqueue.JobManager;
import com.path.android.jobqueue.JobQueue;
+import com.path.android.jobqueue.Params;
import com.path.android.jobqueue.nonPersistentQueue.NonPersistentPriorityQueue;
import com.path.android.jobqueue.test.util.JobQueueFactory;
+import static org.hamcrest.CoreMatchers.*;
+import org.hamcrest.*;
+import org.junit.Test;
import org.junit.runner.RunWith;
-import org.robolectric.RobolectricTestRunner;
+import org.robolectric.*;
@RunWith(RobolectricTestRunner.class)
public class NonPersistentJobQueueTest extends JobQueueTestBase {
@@ -16,4 +22,32 @@ public JobQueue createNew(long sessionId, String id) {
}
});
}
+
+ /**
+ * issue #21 https://github.com/path/android-priority-jobqueue/issues/21
+ */
+ @Test
+ public void testTooManyQueueChanges() throws InterruptedException {
+ JobQueue jobQueue = createNewJobQueue();
+ int limit = 10000;
+ long delayMs = 2000;
+ long then = System.nanoTime() + delayMs * JobManager.NS_PER_MS;
+ for(int i = 0; i < limit; i++) {
+ jobQueue.insert(createNewJobHolder(new Params(0).requireNetwork().delayInMs(delayMs)));
+ }
+
+ MatcherAssert.assertThat("all jobs require network, should return null", jobQueue.nextJobAndIncRunCount(false, null), nullValue());
+ long sleep = then - System.nanoTime();
+ sleep += JobManager.NS_PER_MS * 1000;
+ if (sleep > 0) {
+ Thread.sleep(sleep / JobManager.NS_PER_MS);
+ }
+ //should be able to get it w/o an overflow
+ for(int i = 0; i < limit; i++) {
+ JobHolder holder = jobQueue.nextJobAndIncRunCount(true, null);
+ MatcherAssert.assertThat("should get a next job", holder, notNullValue());
+ jobQueue.remove(holder);
+ }
+
+ }
}