@Test public void testGetNextMessage_moreThreads() throws InterruptedException { TransactionTemplate txTemplate = new TransactionTemplate(jpaTransactionManager); txTemplate.execute(new TransactionCallbackWithoutResult() { @Override protected void doInTransactionWithoutResult(TransactionStatus status) { Message msg = insertNewMessage("1234_4567"); confirmationService.insertFailedConfirmation(msg); msg = insertNewMessage("1234_4567_8"); confirmationService.insertFailedConfirmation(msg); msg = insertNewMessage("1234_4567_9"); confirmationService.insertFailedConfirmation(msg); } }); int threads = 5; final CountDownLatch latch = new CountDownLatch(threads); Runnable task = new Runnable() { @Override public void run() { try { pollExecutor.run(); } finally { latch.countDown(); } } }; mock.expectedMessageCount(3); for (int i = 0; i < threads; i++) { new Thread(task).start(); } latch.await(); mock.assertIsSatisfied(); ExternalCall extCall1 = findConfirmation("1234_4567"); assertThat(extCall1, notNullValue()); assertThat(extCall1.getState(), is(ExternalCallStateEnum.PROCESSING)); ExternalCall extCall2 = findConfirmation("1234_4567_8"); assertThat(extCall2, notNullValue()); assertThat(extCall2.getState(), is(ExternalCallStateEnum.PROCESSING)); ExternalCall extCall3 = findConfirmation("1234_4567_9"); assertThat(extCall3, notNullValue()); assertThat(extCall3.getState(), is(ExternalCallStateEnum.PROCESSING)); confirmationService.confirmationFailed(extCall1); extCall1 = findConfirmation("1234_4567"); assertThat(extCall1.getState(), is(ExternalCallStateEnum.FAILED)); confirmationService.confirmationComplete(extCall2); extCall2 = findConfirmation("1234_4567_8"); assertThat(extCall2.getState(), is(ExternalCallStateEnum.OK)); }
@Override public void run() { Log.debug("Confirmation pooling starts ..."); ExternalCall extCall = null; int lockFailureCount = 0; while (true) { try { extCall = confirmationPool.getNextConfirmation(); if (extCall != null) { producerTemplate.sendBody(targetURI, extCall); } else { break; } } catch (LockFailureException ex) { lockFailureCount++; if (lockFailureCount > LOCK_FAILURE_LIMIT) { Log.warn("Probably problem with locking confirmations - count of lock failures exceeds limit (" + LOCK_FAILURE_LIMIT + ")."); break; } } catch (Exception ex) { Log.error("Error occurred while getting confirmations " + (extCall != null ? extCall.toHumanString() : ""), ex); } } Log.debug("Confirmation pooling finished."); }
ConfirmationPollExecutor implements Runnable { @Override public void run() { Log.debug("Confirmation pooling starts ..."); ExternalCall extCall = null; int lockFailureCount = 0; while (true) { try { extCall = confirmationPool.getNextConfirmation(); if (extCall != null) { producerTemplate.sendBody(targetURI, extCall); } else { break; } } catch (LockFailureException ex) { lockFailureCount++; if (lockFailureCount > LOCK_FAILURE_LIMIT) { Log.warn("Probably problem with locking confirmations - count of lock failures exceeds limit (" + LOCK_FAILURE_LIMIT + ")."); break; } } catch (Exception ex) { Log.error("Error occurred while getting confirmations " + (extCall != null ? extCall.toHumanString() : ""), ex); } } Log.debug("Confirmation pooling finished."); } }
ConfirmationPollExecutor implements Runnable { @Override public void run() { Log.debug("Confirmation pooling starts ..."); ExternalCall extCall = null; int lockFailureCount = 0; while (true) { try { extCall = confirmationPool.getNextConfirmation(); if (extCall != null) { producerTemplate.sendBody(targetURI, extCall); } else { break; } } catch (LockFailureException ex) { lockFailureCount++; if (lockFailureCount > LOCK_FAILURE_LIMIT) { Log.warn("Probably problem with locking confirmations - count of lock failures exceeds limit (" + LOCK_FAILURE_LIMIT + ")."); break; } } catch (Exception ex) { Log.error("Error occurred while getting confirmations " + (extCall != null ? extCall.toHumanString() : ""), ex); } } Log.debug("Confirmation pooling finished."); }  }
ConfirmationPollExecutor implements Runnable { @Override public void run() { Log.debug("Confirmation pooling starts ..."); ExternalCall extCall = null; int lockFailureCount = 0; while (true) { try { extCall = confirmationPool.getNextConfirmation(); if (extCall != null) { producerTemplate.sendBody(targetURI, extCall); } else { break; } } catch (LockFailureException ex) { lockFailureCount++; if (lockFailureCount > LOCK_FAILURE_LIMIT) { Log.warn("Probably problem with locking confirmations - count of lock failures exceeds limit (" + LOCK_FAILURE_LIMIT + ")."); break; } } catch (Exception ex) { Log.error("Error occurred while getting confirmations " + (extCall != null ? extCall.toHumanString() : ""), ex); } } Log.debug("Confirmation pooling finished."); }  @Override void run(); }
ConfirmationPollExecutor implements Runnable { @Override public void run() { Log.debug("Confirmation pooling starts ..."); ExternalCall extCall = null; int lockFailureCount = 0; while (true) { try { extCall = confirmationPool.getNextConfirmation(); if (extCall != null) { producerTemplate.sendBody(targetURI, extCall); } else { break; } } catch (LockFailureException ex) { lockFailureCount++; if (lockFailureCount > LOCK_FAILURE_LIMIT) { Log.warn("Probably problem with locking confirmations - count of lock failures exceeds limit (" + LOCK_FAILURE_LIMIT + ")."); break; } } catch (Exception ex) { Log.error("Error occurred while getting confirmations " + (extCall != null ? extCall.toHumanString() : ""), ex); } } Log.debug("Confirmation pooling finished."); }  @Override void run();  }