@Test public void shouldExecuteSagaRolledBack() { initializeSagaManager(); startSaga(); reset(sagaInstanceRepository, sagaCommandProducer); handleReply(true); verify(testSaga).onSagaRolledBack(eq(sagaId), any(TestSagaData.class)); reset(sagaInstanceRepository, sagaCommandProducer); }
private void handleReply(Message message) { if (!isReplyForThisSagaType(message)) return; logger.debug("Handle reply: {}", message); String sagaId = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_ID); String sagaType = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_TYPE); SagaInstance sagaInstance = sagaInstanceRepository.find(sagaType, sagaId); Data sagaData = SagaDataSerde.deserializeSagaData(sagaInstance.getSerializedSagaData()); message.getHeader(SagaReplyHeaders.REPLY_LOCKED).ifPresent(lockedTarget -> { String destination = message.getRequiredHeader(CommandMessageHeaders.inReply(CommandMessageHeaders.DESTINATION)); sagaInstance.addDestinationsAndResources(singleton(new DestinationAndResource(destination, lockedTarget))); }); String currentState = sagaInstance.getStateName(); logger.info("Current state={}", currentState); SagaActions<Data> actions = getStateDefinition().handleReply(currentState, sagaData, message); logger.info("Handled reply. Sending commands {}", actions.getCommands()); processActions(sagaId, sagaInstance, sagaData, actions); }
SagaManagerImpl implements SagaManager<Data> { private void handleReply(Message message) { if (!isReplyForThisSagaType(message)) return; logger.debug("Handle reply: {}", message); String sagaId = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_ID); String sagaType = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_TYPE); SagaInstance sagaInstance = sagaInstanceRepository.find(sagaType, sagaId); Data sagaData = SagaDataSerde.deserializeSagaData(sagaInstance.getSerializedSagaData()); message.getHeader(SagaReplyHeaders.REPLY_LOCKED).ifPresent(lockedTarget -> { String destination = message.getRequiredHeader(CommandMessageHeaders.inReply(CommandMessageHeaders.DESTINATION)); sagaInstance.addDestinationsAndResources(singleton(new DestinationAndResource(destination, lockedTarget))); }); String currentState = sagaInstance.getStateName(); logger.info("Current state={}", currentState); SagaActions<Data> actions = getStateDefinition().handleReply(currentState, sagaData, message); logger.info("Handled reply. Sending commands {}", actions.getCommands()); processActions(sagaId, sagaInstance, sagaData, actions); } }
SagaManagerImpl implements SagaManager<Data> { private void handleReply(Message message) { if (!isReplyForThisSagaType(message)) return; logger.debug("Handle reply: {}", message); String sagaId = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_ID); String sagaType = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_TYPE); SagaInstance sagaInstance = sagaInstanceRepository.find(sagaType, sagaId); Data sagaData = SagaDataSerde.deserializeSagaData(sagaInstance.getSerializedSagaData()); message.getHeader(SagaReplyHeaders.REPLY_LOCKED).ifPresent(lockedTarget -> { String destination = message.getRequiredHeader(CommandMessageHeaders.inReply(CommandMessageHeaders.DESTINATION)); sagaInstance.addDestinationsAndResources(singleton(new DestinationAndResource(destination, lockedTarget))); }); String currentState = sagaInstance.getStateName(); logger.info("Current state={}", currentState); SagaActions<Data> actions = getStateDefinition().handleReply(currentState, sagaData, message); logger.info("Handled reply. Sending commands {}", actions.getCommands()); processActions(sagaId, sagaInstance, sagaData, actions); } SagaManagerImpl(Saga<Data> saga,
                         SagaInstanceRepository sagaInstanceRepository,
                         CommandProducer commandProducer,
                         MessageConsumer messageConsumer,
                         SagaLockManager sagaLockManager,
                         SagaCommandProducer sagaCommandProducer); }
SagaManagerImpl implements SagaManager<Data> { private void handleReply(Message message) { if (!isReplyForThisSagaType(message)) return; logger.debug("Handle reply: {}", message); String sagaId = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_ID); String sagaType = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_TYPE); SagaInstance sagaInstance = sagaInstanceRepository.find(sagaType, sagaId); Data sagaData = SagaDataSerde.deserializeSagaData(sagaInstance.getSerializedSagaData()); message.getHeader(SagaReplyHeaders.REPLY_LOCKED).ifPresent(lockedTarget -> { String destination = message.getRequiredHeader(CommandMessageHeaders.inReply(CommandMessageHeaders.DESTINATION)); sagaInstance.addDestinationsAndResources(singleton(new DestinationAndResource(destination, lockedTarget))); }); String currentState = sagaInstance.getStateName(); logger.info("Current state={}", currentState); SagaActions<Data> actions = getStateDefinition().handleReply(currentState, sagaData, message); logger.info("Handled reply. Sending commands {}", actions.getCommands()); processActions(sagaId, sagaInstance, sagaData, actions); } SagaManagerImpl(Saga<Data> saga,
                         SagaInstanceRepository sagaInstanceRepository,
                         CommandProducer commandProducer,
                         MessageConsumer messageConsumer,
                         SagaLockManager sagaLockManager,
                         SagaCommandProducer sagaCommandProducer); void setSagaCommandProducer(SagaCommandProducer sagaCommandProducer); void setSagaInstanceRepository(SagaInstanceRepository sagaInstanceRepository); void setCommandProducer(CommandProducer commandProducer); void setMessageConsumer(MessageConsumer messageConsumer); void setSagaLockManager(SagaLockManager sagaLockManager); @Override SagaInstance create(Data sagaData); @Override SagaInstance create(Data data, Class targetClass, Object targetId); @Override SagaInstance create(Data sagaData, Optional<String> resource); @PostConstruct void subscribeToReplyChannel(); void handleMessage(Message message); }
SagaManagerImpl implements SagaManager<Data> { private void handleReply(Message message) { if (!isReplyForThisSagaType(message)) return; logger.debug("Handle reply: {}", message); String sagaId = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_ID); String sagaType = message.getRequiredHeader(SagaReplyHeaders.REPLY_SAGA_TYPE); SagaInstance sagaInstance = sagaInstanceRepository.find(sagaType, sagaId); Data sagaData = SagaDataSerde.deserializeSagaData(sagaInstance.getSerializedSagaData()); message.getHeader(SagaReplyHeaders.REPLY_LOCKED).ifPresent(lockedTarget -> { String destination = message.getRequiredHeader(CommandMessageHeaders.inReply(CommandMessageHeaders.DESTINATION)); sagaInstance.addDestinationsAndResources(singleton(new DestinationAndResource(destination, lockedTarget))); }); String currentState = sagaInstance.getStateName(); logger.info("Current state={}", currentState); SagaActions<Data> actions = getStateDefinition().handleReply(currentState, sagaData, message); logger.info("Handled reply. Sending commands {}", actions.getCommands()); processActions(sagaId, sagaInstance, sagaData, actions); } SagaManagerImpl(Saga<Data> saga,
                         SagaInstanceRepository sagaInstanceRepository,
                         CommandProducer commandProducer,
                         MessageConsumer messageConsumer,
                         SagaLockManager sagaLockManager,
                         SagaCommandProducer sagaCommandProducer); void setSagaCommandProducer(SagaCommandProducer sagaCommandProducer); void setSagaInstanceRepository(SagaInstanceRepository sagaInstanceRepository); void setCommandProducer(CommandProducer commandProducer); void setMessageConsumer(MessageConsumer messageConsumer); void setSagaLockManager(SagaLockManager sagaLockManager); @Override SagaInstance create(Data sagaData); @Override SagaInstance create(Data data, Class targetClass, Object targetId); @Override SagaInstance create(Data sagaData, Optional<String> resource); @PostConstruct void subscribeToReplyChannel(); void handleMessage(Message message);  }