/**
* Ordered Queue Test
*/
@Test(timeout = DEFAULT_TEST_TIMEOUT)
public void testOrderedQueue() throws Exception {
final JobManager jobManager = this.getJobManager();
// register consumer and event handler
final Barrier cb = new Barrier(2);
final AtomicInteger count = new AtomicInteger(0);
final AtomicInteger parallelCount = new AtomicInteger(0);
final ServiceRegistration jcReg = this.registerJobConsumer("sling/orderedtest/*",
new JobConsumer() {
private volatile int lastCounter = -1;
@Override
public JobResult process(final Job job) {
final int counter = job.getProperty("counter", -10);
assertNotEquals("Counter property is missing", -10, counter);
assertTrue("Counter should only increment by max of 1 " + counter + " - " + lastCounter,
counter == lastCounter || counter == lastCounter +1);
lastCounter = counter;
if ("sling/orderedtest/start".equals(job.getTopic()) ) {
cb.block();
return JobResult.OK;
}
if ( parallelCount.incrementAndGet() > 1 ) {
parallelCount.decrementAndGet();
return JobResult.FAILED;
}
final String topic = job.getTopic();
if ( topic.endsWith("sub1") ) {
final int i = (Integer)job.getProperty(Job.PROPERTY_JOB_RETRY_COUNT);
if ( i == 0 ) {
parallelCount.decrementAndGet();
return JobResult.FAILED;
}
}
try {
Thread.sleep(30);
} catch (InterruptedException ie) {
// ignore
}
parallelCount.decrementAndGet();
return JobResult.OK;
}
});
final ServiceRegistration ehReg = this.registerEventHandler(NotificationConstants.TOPIC_JOB_FINISHED,
new EventHandler() {
@Override
public void handleEvent(final Event event) {
count.incrementAndGet();
}
});
try {
// we first sent one event to get the queue started
final Map<String, Object> properties = new HashMap<String, Object>();
properties.put("counter", -1);
jobManager.addJob("sling/orderedtest/start", properties);
assertTrue("No event received in the given time.", cb.block(5));
cb.reset();
// get the queue
final Queue q = jobManager.getQueue("orderedtest");
assertNotNull("Queue 'orderedtest' should exist!", q);
// suspend it
q.suspend();
final int NUM_JOBS = 30;
// we start "some" jobs:
for(int i = 0; i < NUM_JOBS; i++ ) {
final String subTopic = "sling/orderedtest/sub" + (i % 10);
properties.clear();
properties.put("counter", i);
jobManager.addJob(subTopic, properties);
}
// start the queue
q.resume();
while ( count.get() < NUM_JOBS +1 ) {
try {
Thread.sleep(500);
} catch (InterruptedException ie) {
// ignore
}
}
// we started one event before the test, so add one
assertEquals("Finished count", NUM_JOBS + 1, count.get());
assertEquals("Finished count", NUM_JOBS + 1, jobManager.getStatistics().getNumberOfFinishedJobs());
assertEquals("Finished count", NUM_JOBS + 1, q.getStatistics().getNumberOfFinishedJobs());
assertEquals("Failed count", NUM_JOBS / 10, q.getStatistics().getNumberOfFailedJobs());
assertEquals("Cancelled count", 0, q.getStatistics().getNumberOfCancelledJobs());
} finally {
jcReg.unregister();