ThreadPoolCallbackService implements CallbackService { @Override public void run(Runnable runnable, Object orderingCtx, Promise<Void> promise) { final String methodName = "run"; logger.entry(this, methodName, runnable, orderingCtx, promise); int hash = orderingCtx.hashCode(); if (hash == Integer.MIN_VALUE) hash = Integer.MIN_VALUE + 1; workLists[Math.abs(hash) % poolSize].put(runnable, promise); logger.exit(this, methodName); } ThreadPoolCallbackService(int poolSize); @Override void run(Runnable runnable, Object orderingCtx, Promise<Void> promise);  }
@Test public void successfulCallback() throws InterruptedException { CallbackService cbs = new ThreadPoolCallbackService(5); MockCallbackPromise promise = new MockCallbackPromise(Method.SUCCESS, false); final AtomicBoolean run = new AtomicBoolean(false); cbs.run(new Runnable() { @Override public void run() { run.set(true); } }, new Object(), promise); assertTrue("Promise should have been completed", promise.waitForComplete(2500)); assertTrue("Promise should have been completed successfully", promise.isSuccessful()); assertTrue("Runnable should have been run!", run.get()); }
@Test public void exceptionThrownInCallback() throws InterruptedException { CallbackService cbs = new ThreadPoolCallbackService(5); MockCallbackPromise promise = new MockCallbackPromise(Method.FAILURE, false); final RuntimeException exception = new RuntimeException(); cbs.run(new Runnable() { @Override public void run() { throw exception; } }, new Object(), promise); assertTrue("Promise should have been completed", promise.waitForComplete(2500)); assertTrue("Promise should not have been completed successfully", !promise.isSuccessful()); assertSame("Exception should have been thrown from run()!", exception, promise.getException()); }
@Test public void errorThrownInCallback() throws InterruptedException { CallbackService cbs = new ThreadPoolCallbackService(5); MockCallbackPromise promise = new MockCallbackPromise(Method.FAILURE, false); final Error error = new AssertionError(); cbs.run(new Runnable() { @Override public void run() { throw error; } }, new Object(), promise); assertTrue("Promise should have been completed", promise.waitForComplete(2500)); assertTrue("Promise should not have been completed successfully", !promise.isSuccessful()); assertSame("Error should have been thrown from run()!", error, promise.getException().getCause()); }
@Test public void stress() throws InterruptedException { final int callbacks = 10000; final int orderingContexts = 50; final int poolSize = 5; final ThreadPoolCallbackService callbackService = new ThreadPoolCallbackService(poolSize); class PromiseList extends LinkedList<StressCallbackPromise> { private static final long serialVersionUID = 2857979802052992226L; } PromiseList[] promiseList = new PromiseList[orderingContexts]; Object[] contexts = new Object[orderingContexts]; for (int i = 0; i < orderingContexts; ++i) { promiseList[i] = new PromiseList(); contexts[i] = new Object(); } ArrayList<MockCallbackPromise> allPromises = new ArrayList<>(callbacks); Random random = new Random(); for (int i = 0; i < callbacks; ++i) { Method expectedOutcome = Math.abs(random.nextInt() % 100) > 5 ? Method.SUCCESS : Method.FAILURE; StressCallbackPromise promise = new StressCallbackPromise(expectedOutcome); int x = Math.abs(random.nextInt() % orderingContexts); PromiseList list = promiseList[x]; allPromises.add(promise); list.addLast(promise); callbackService.run(promise.getRunnable(), contexts[x], promise); } int count = 0; while(!allPromises.isEmpty()) { MockCallbackPromise p = allPromises.remove(0); assertTrue("Expected promise #" + count + " to have been completed", p.waitForComplete(1000)); if (p.getExpectedMethod() == Method.SUCCESS) { assertTrue("Promise #" + count + " should have been completed successfully", p.isSuccessful()); } else { assertTrue("Promise #" + count + " should have been completed unsuccessfully", !p.isSuccessful()); } ++count; } for (int i = 0; i < orderingContexts; ++i) { if (promiseList[i].size() > 1) { int n = promiseList[i].removeFirst().getSequence(); count = 0; while(!promiseList[i].isEmpty()) { int m = promiseList[i].removeFirst().getSequence(); assertTrue("Out of order promise completion: i=" + i + " count=" + count + " n=" + n + " m=" + m, n < m); n = m; ++count; } } } }