@Test public void testRepairProcessingMessagesToFailed() throws Exception { setPrivateField(messageService, "countPartlyFailsBeforeFailed", 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 MessageProcessor() { @Override public void process(Message message) { message.setState(MsgStateEnum.PROCESSING); message.setFailedCount(2); message.setMsgTimestamp(DateTime.now().minusHours(1).toDate()); message.setStartProcessTimestamp(DateTime.now().minusHours(1).toDate()); message.setLastUpdateTimestamp(DateTime.now().minusHours(1).toDate()); } })[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)); }
@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; } }
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; } } }
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; } }  }
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; } }  @Override void repairProcessingMessages(); @Required void setTransactionManager(JpaTransactionManager transactionManager); }
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; } }  @Override void repairProcessingMessages(); @Required void setTransactionManager(JpaTransactionManager transactionManager);  }