RepairMessageServiceDbImpl implements RepairMessageService { @Override public void repairProcessingMessages() { List<Message> messages = findProcessingMessages(); LOG.debug("Found {} message(s) for repairing ...", messages.size()); int batchStartIncl = 0; int batchEndExcl; while (batchStartIncl < messages.size()) { batchEndExcl = min(batchStartIncl + BATCH_SIZE, messages.size()); updateMessagesInDB(messages.subList(batchStartIncl, batchEndExcl)); batchStartIncl = batchEndExcl; } } @Autowired  RepairMessageServiceDbImpl(PlatformTransactionManager transactionManager); @Override void repairProcessingMessages();  }
@Test public void testRepairProcessingMessagesMany() throws Exception { List<Message> messages = new LinkedList<>(); messages.addAll(Arrays.asList(createAndSaveMessages(70, MsgStateEnum.PROCESSING))); messages.addAll(Arrays.asList(createAndSaveMessages(30, MsgStateEnum.NEW))); messages.addAll(Arrays.asList(createAndSaveMessages(28, MsgStateEnum.IN_QUEUE))); messageService.repairProcessingMessages(); messageService.repairProcessingMessages(); messageService.repairProcessingMessages(); for (Message message : messages) { LOG.info("Verifying message {}", message); Message found = messageDao.findMessage(message.getMsgId()); assertThat(found, notNullValue()); assertThat(found.getState(), is(MsgStateEnum.PARTLY_FAILED)); } }
@Test public void testRepairProcessingMessagesToFailed() throws Exception { setPrivateField(messageService, "countPartlyFailsBeforeFailed", new FixedConfigurationItem<>(2)); RouteBuilder failedRoute = new RouteBuilder() { @Override public void configure() throws Exception { from(AsynchConstants.URI_ERROR_FATAL) .to("mock:test"); } }; getCamelContext().addRoutes(failedRoute); MockEndpoint failedMock = getCamelContext().getEndpoint("mock:test", MockEndpoint.class); failedMock.expectedMessageCount(1); Message message = createAndSaveMessages(1, new MessageCallback() { @Override public void beforeInsert(Message message, int order) { message.setState(MsgStateEnum.PROCESSING); message.setFailedCount(2); message.setMsgTimestamp(Instant.now().minus(1, ChronoUnit.HOURS)); message.setStartProcessTimestamp(Instant.now().minus(1, ChronoUnit.HOURS)); message.setLastUpdateTimestamp(Instant.now().minus(1, ChronoUnit.HOURS)); } })[0]; messageService.repairProcessingMessages(); failedMock.assertIsSatisfied(); Message found = messageDao.findMessage(message.getMsgId()); assertThat(found, notNullValue()); assertThat(found.getState(), is(MsgStateEnum.PROCESSING)); Exchange exchange = failedMock.getExchanges().get(0); assertThat(exchange.getIn().getBody(), nullValue()); assertThat(exchange.getProperty(Exchange.EXCEPTION_CAUGHT), notNullValue()); assertThat(exchange.getProperty(Exchange.EXCEPTION_CAUGHT), instanceOf(IntegrationException.class)); assertThat((Message)exchange.getIn().getHeader(AsynchConstants.MSG_HEADER), is(found)); }
@Test public void testRepairProcessingMessagesOne() throws Exception { Message message = createAndSaveMessages(1, MsgStateEnum.PROCESSING)[0]; messageService.repairProcessingMessages(); Message found = messageDao.findMessage(message.getMsgId()); assertThat(found, notNullValue()); assertThat(found.getState(), is(MsgStateEnum.PARTLY_FAILED)); }