@Test @Transactional public void testGuaranteedOrder_processing_differentFunnelValues() throws InterruptedException { insertNewMessage("id1", MsgStateEnum.PROCESSING, true, "someValue"); Message msg = insertNewMessage("id2", MsgStateEnum.PROCESSING, true, FUNNEL_VALUE); messagePollExecutor.startMessageProcessing(msg); Assert.assertThat(em.find(Message.class, msg.getMsgId()).getState(), CoreMatchers.is(MsgStateEnum.PROCESSING)); }
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 { Date failedDate = DateUtils.addSeconds(new Date(), -postponedIntervalWhenFailed); final Message paramMsg = msg; if (msg.getReceiveTimestamp().before(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); } } }
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 { Date failedDate = DateUtils.addSeconds(new Date(), -postponedIntervalWhenFailed); final Message paramMsg = msg; if (msg.getReceiveTimestamp().before(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); } } } }
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 { Date failedDate = DateUtils.addSeconds(new Date(), -postponedIntervalWhenFailed); final Message paramMsg = msg; if (msg.getReceiveTimestamp().before(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); } } }  }
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 { Date failedDate = DateUtils.addSeconds(new Date(), -postponedIntervalWhenFailed); final Message paramMsg = msg; if (msg.getReceiveTimestamp().before(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(); }
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 { Date failedDate = DateUtils.addSeconds(new Date(), -postponedIntervalWhenFailed); final Message paramMsg = msg; if (msg.getReceiveTimestamp().before(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();  }