FinalMessagesProcessingServiceImpl implements FinalMessagesProcessingService { @Override public void processMessages() { Assert.notNull(finalMessageStatesConfig, "finalMessageStatesConfig must not be null"); for (Map.Entry<MsgStateEnum, String> msgStateEntry : finalMessageStatesConfig.entrySet()) { final List<Message> messageList = findMessagesForProcessing(msgStateEntry.getKey()); LOG.trace("Will process '{}' messages in state '{}'.", messageList.size(), msgStateEntry.getKey()); for (Message message : messageList) { LogContextHelper.setLogContextParams(message, null); try { transactionTemplate.execute((TransactionStatus status) -> { for (FinalMessageProcessor finalMessageProcessor : finalMessageProcessorList) { finalMessageProcessor.processMessage(message); } return null; }); } catch (Exception ex) { LOG.error("Failed to process message : {}, will continue with next.", message, ex); } LogContextHelper.removeLogContextParams(); } } } @Autowired  FinalMessagesProcessingServiceImpl(PlatformTransactionManager transactionManager); @EventListener void onApplicationEvent(ApplicationReadyEvent event); @Override void processMessages();  }
@Test public void test_timing() { createAndSaveMessages(3, (message, order) -> { switch (order) { case 1: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec + 1)); break; case 2: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec)); break; case 3: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now()); break; } }); messagesProcessingService.processMessages(); verify(finalMessageProcessor, times(2)).processMessage(any(Message.class)); verify(additionalMessageProcessor, times(2)).processMessage(any(Message.class)); }
@Test public void test_split() { createAndSaveMessages(2, (message, order) -> { switch (order) { case 1: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec + 1)); break; case 2: message.setState(MsgStateEnum.FAILED); message.setLastUpdateTimestamp(Instant.now().minusSeconds(failedSaveTimeInSec + 1)); break; } }); messagesProcessingService.processMessages(); verify(finalMessageProcessor, times(2)).processMessage(any(Message.class)); verify(additionalMessageProcessor, times(2)).processMessage(any(Message.class)); }
@Test public void test_state_ignoring() { createAndSaveMessages(5, (message, order) -> { switch (order) { case 1: message.setState(MsgStateEnum.NEW); break; case 2: message.setState(MsgStateEnum.IN_QUEUE); break; case 3: message.setState(MsgStateEnum.PROCESSING); break; case 4: message.setState(MsgStateEnum.WAITING); break; case 5: message.setState(MsgStateEnum.PARTLY_FAILED); break; } }); messagesProcessingService.processMessages(); }
@Test public void test_technicalCountLimit() { ReflectionTestUtils.setField(messagesProcessingService, "messagesTechnicalLimit", new FixedConfigurationItem<>(2L)); createAndSaveMessages(3, (message, order) -> { switch (order) { case 1: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec + 1)); break; case 2: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec + 1)); break; case 3: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec + 1)); break; } }); messagesProcessingService.processMessages(); verify(finalMessageProcessor, times(2)).processMessage(any(Message.class)); }
@Test public void test_errorsAreSkipped() { createAndSaveMessages(3, (message, order) -> { switch (order) { case 1: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec + 1)); break; case 2: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec + 1)); break; case 3: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec + 1)); break; } }); doThrow(new RuntimeException("General exception")).when(finalMessageProcessor).processMessage(any(Message.class)); messagesProcessingService.processMessages(); verify(finalMessageProcessor, times(3)).processMessage(any(Message.class)); }
@Test public void test_processorOrder() { when(finalMessageProcessor.getOrder()).thenReturn(0); when(additionalMessageProcessor.getOrder()).thenReturn(1); createAndSaveMessages(1, (message, order) -> { switch (order) { case 1: message.setState(MsgStateEnum.OK); message.setLastUpdateTimestamp(Instant.now().minusSeconds(okSaveTimeInSec + 1)); break; } }); messagesProcessingService.processMessages(); InOrder inOrder = inOrder(finalMessageProcessor, additionalMessageProcessor); inOrder.verify(finalMessageProcessor, times(1)).processMessage(any(Message.class)); inOrder.verify(additionalMessageProcessor, times(1)).processMessage(any(Message.class)); }
@Test public void test_keepIndefinitely() { createAndSaveMessages(2, (message, order) -> { switch (order) { case 1: message.setState(MsgStateEnum.CANCEL); message.setLastUpdateTimestamp(Instant.now().minusSeconds(1_000_000)); break; case 2: message.setState(MsgStateEnum.CANCEL); message.setLastUpdateTimestamp(Instant.now().minusSeconds(42)); break; } }); messagesProcessingService.processMessages(); verify(finalMessageProcessor, times(0)).processMessage(any(Message.class)); verify(additionalMessageProcessor, times(0)).processMessage(any(Message.class)); }