MessagePollExecutor implements Runnable { void startMessageProcessing(Message msg) { Assert.notNull(msg, "the msg must not be null"); if (isMsgInGuaranteedOrder(msg)) { producerTemplate.sendBodyAndHeader(targetURI, ExchangePattern.InOnly, msg, AsynchConstants.MSG_QUEUE_INSERT_HEADER, System.currentTimeMillis()); } else { Instant failedDate = Instant.now().minusSeconds(postponedIntervalWhenFailed.getValue().getSeconds()); final Message paramMsg = msg; if (msg.getReceiveTimestamp().isBefore(failedDate)) { producerTemplate.send(AsynchConstants.URI_ERROR_FATAL, ExchangePattern.InOnly, new Processor() { @Override public void process(Exchange exchange) throws Exception { IntegrationException ex = new IntegrationException(InternalErrorEnum.E121, "Message " + paramMsg.toHumanString() + " exceeded interval for starting " + "processing => changed to FAILED state"); exchange.setProperty(Exchange.EXCEPTION_CAUGHT, ex); exchange.getIn().setHeader(AsynchConstants.MSG_HEADER, paramMsg); } }); } else { messageService.setStatePostponed(msg); ExchangeBuilder exchangeBuilder = ExchangeBuilder.anExchange(producerTemplate.getCamelContext()); Exchange exchange = exchangeBuilder.build(); exchange.getIn().setHeader(AsynchConstants.MSG_HEADER, paramMsg); AsynchEventHelper.notifyMsgPostponed(exchange); } } }  @Override void run();  }
@Test @Transactional public void testGuaranteedOrder_postponedMessage() throws InterruptedException { insertNewMessage("id1", MsgStateEnum.PROCESSING, FUNNEL_VALUE, true); Message msg = insertNewMessage("id2", MsgStateEnum.PROCESSING, FUNNEL_VALUE, true); msg.setReceiveTimestamp(Instant.now().plusSeconds(10)); messagePollExecutor.startMessageProcessing(msg); Assert.assertThat(em.find(Message.class, msg.getMsgId()).getState(), CoreMatchers.is(MsgStateEnum.POSTPONED)); }
@Test @Transactional public void testGuaranteedOrder_processing_onlyOneMessage() throws InterruptedException { Message msg = insertNewMessage("id2", MsgStateEnum.PROCESSING, FUNNEL_VALUE, true); messagePollExecutor.startMessageProcessing(msg); Assert.assertThat(em.find(Message.class, msg.getMsgId()).getState(), CoreMatchers.is(MsgStateEnum.PROCESSING)); }
@Test @Transactional public void testGuaranteedOrder_processing_msgIsNotGuaranteed() throws InterruptedException { insertNewMessage("id1", MsgStateEnum.PROCESSING, FUNNEL_VALUE, false); Message msg = insertNewMessage("id2", MsgStateEnum.PROCESSING, FUNNEL_VALUE, true); messagePollExecutor.startMessageProcessing(msg); Assert.assertThat(em.find(Message.class, msg.getMsgId()).getState(), CoreMatchers.is(MsgStateEnum.PROCESSING)); }
@Test @Transactional public void testGuaranteedOrder_processing_differentFunnelValues() throws InterruptedException { insertNewMessage("id1", MsgStateEnum.PROCESSING, "someValue", true); Message msg = insertNewMessage("id2", MsgStateEnum.PROCESSING, FUNNEL_VALUE, true); messagePollExecutor.startMessageProcessing(msg); Assert.assertThat(em.find(Message.class, msg.getMsgId()).getState(), CoreMatchers.is(MsgStateEnum.PROCESSING)); }
@Test @Transactional public void testGuaranteedOrder_failedMessage() throws InterruptedException { insertNewMessage("id1", MsgStateEnum.PROCESSING, FUNNEL_VALUE, true); Message msg = insertNewMessage("id2", MsgStateEnum.PROCESSING, FUNNEL_VALUE, true); msg.setReceiveTimestamp(Instant.now().minusSeconds(10)); messagePollExecutor.startMessageProcessing(msg); Assert.assertThat(em.find(Message.class, msg.getMsgId()).getState(), CoreMatchers.is(MsgStateEnum.FAILED)); }