SagaExecutionTask implements ExecutedRunnable { @Override public void run() { try { handle(); } catch (Exception e) { Throwables.throwIfUnchecked(e); throw new RuntimeException(e); } } @Inject  SagaExecutionTask(
            final SagaEnvironment environment,
            final HandlerInvoker invoker,
            final Object message,
            final Map<HeaderName<?>, Object> headers,
            @Nullable final ExecutionContext parentContext); void handle(); @Override void run(); @Override Object message(); @Override Optional<T> getHeaderValue(final HeaderName<T> header);  }
@Test public void run_sagaNotCompleted_saveSagaState() { when(saga.isFinished()).thenReturn(false); sut.run(); verify(storage).save(state); }
@Test public void run_sagaNotCompleted_rememberStoredIdInContext() { final String sagaId = RandomStringUtils.randomAlphanumeric(10); when(state.getSagaId()).thenReturn(sagaId); when(saga.isFinished()).thenReturn(false); sut.run(); verify(context).recordSagaStateStored(sagaId); }
@Test public void run_sagaIsStartedAndFinished_doesNotSaveSate() { when(saga.isFinished()).thenReturn(true); when(sagaInstanceInfo.isStarting()).thenReturn(true); sut.run(); verify(storage, never()).save(any(SagaState.class)); }
@Test public void run_sagaIsStartedAndFinished_doesNotDeleteSate() { when(saga.isFinished()).thenReturn(true); when(sagaInstanceInfo.isStarting()).thenReturn(true); sut.run(); verify(storage, never()).delete(any(String.class)); }
@Test public void run_sagaHasContinuedAndIsFinished_deleteSagaState() { final String sagaId = RandomStringUtils.randomAlphanumeric(10); when(state.getSagaId()).thenReturn(sagaId); when(saga.isFinished()).thenReturn(true); when(sagaInstanceInfo.isStarting()).thenReturn(false); sut.run(); verify(storage).delete(sagaId); }
@Test public void run_sagaHasContinuedAndIsFinished_cancelTimeouts() { final String sagaId = RandomStringUtils.randomAlphanumeric(10); when(state.getSagaId()).thenReturn(sagaId); when(saga.isFinished()).thenReturn(true); when(sagaInstanceInfo.isStarting()).thenReturn(false); sut.run(); verify(timeoutManager).cancelTimeouts(sagaId); }
@Test public void run_sagaIsStartedAndIsFinished_doNotCancelTimeouts() { when(saga.isFinished()).thenReturn(true); when(sagaInstanceInfo.isStarting()).thenReturn(true); sut.run(); verify(timeoutManager, never()).cancelTimeouts(any(String.class)); }
@Test public void run_sagaIsStartedAndIsFinishedAndAlreadyStore_deleteState() { final String sagaId = RandomStringUtils.randomAlphanumeric(10); when(saga.isFinished()).thenReturn(true); when(state.getSagaId()).thenReturn(sagaId); when(sagaInstanceInfo.isStarting()).thenReturn(true); when(context.hasBeenStored(saga.state().getSagaId())).thenReturn(true); sut.run(); verify(storage).delete(sagaId); }
@Test public void run_sagaIsStartedAndIsFinishedAndAlreadyStore_cancelTimeout() { final String sagaId = RandomStringUtils.randomAlphanumeric(10); when(saga.isFinished()).thenReturn(true); when(state.getSagaId()).thenReturn(sagaId); when(sagaInstanceInfo.isStarting()).thenReturn(true); when(context.hasBeenStored(saga.state().getSagaId())).thenReturn(true); sut.run(); verify(timeoutManager).cancelTimeouts(sagaId); }
@Test public void run_messageParam_messageIsSetOnContext() throws InvocationTargetException, IllegalAccessException { sut.run(); InOrder inOrder = inOrder(context, invoker); inOrder.verify(context).setMessage(theMessage); inOrder.verify(invoker).invoke(isA(InvocationContext.class)); }
@Test public void run_sagaToInvoke_sagaIsSetOnContext() throws InvocationTargetException, IllegalAccessException { sut.run(); InOrder inOrder = inOrder(context, invoker); inOrder.verify(context).setSaga(saga); inOrder.verify(invoker).invoke(isA(InvocationContext.class)); }
@Test public void run_parentContextProvided_contextHasParentContext() { CurrentExecutionContext context = new SagaExecutionContext(); ExecutionContext parentContext = mock(ExecutionContext.class); SagaEnvironment env = SagaEnvironment.create(timeoutManager, storage, createContextProvider(context), Sets.newHashSet(module), Sets.newHashSet(interceptor), instanceResolver, DefaultModuleCoordinator::new); sut = new SagaExecutionTask(env, invoker, theMessage, Collections.EMPTY_MAP, parentContext); sut.run(); assertThat("Expected header value to be part of context.", context.parentContext(), sameInstance(parentContext)); }
@Test public void run_usingModule_moduleStartedBeforeSaga() throws InvocationTargetException, IllegalAccessException { sut.run(); InOrder inOrder = inOrder(invoker, module); inOrder.verify(module).onStart(context); inOrder.verify(invoker).invoke(isA(InvocationContext.class)); }
@Test public void run_usingModule_moduleFinishedAfterSaga() throws InvocationTargetException, IllegalAccessException { sut.run(); InOrder inOrder = inOrder(invoker, module); inOrder.verify(invoker).invoke(isA(InvocationContext.class)); inOrder.verify(module).onFinished(context); }
@Test public void run_invokeThrows_moduleFinishedGetsCalled() throws InvocationTargetException, IllegalAccessException { doThrow(NullPointerException.class).when(invoker).invoke(saga, theMessage); try { sut.run(); } catch (NullPointerException ex) { } verify(module).onFinished(context); }
@Test public void run_moduleStartThrows_moduleFinishedGetsCalled() throws Exception { doThrow(NullPointerException.class).when(module).onStart(context); try { sut.run(); } catch (NullPointerException ex) { } verify(module).onFinished(context); }
@Test public void run_moduleStartThrows_moduleErrorGetsCalled() throws Exception { NullPointerException expected = new NullPointerException(); doThrow(expected).when(module).onStart(context); try { sut.run(); } catch (NullPointerException ex) { } verify(module).onError(context, theMessage, expected); }
@Test public void run_invokeThrows_moduleErrorGetsCalled() throws InvocationTargetException, IllegalAccessException { NullPointerException npe = new NullPointerException(); doThrow(npe).when(invoker).invoke(isA(InvocationContext.class)); try { sut.run(); } catch (NullPointerException ex) { } verify(module).onError(context, theMessage, npe); }
@Test public void run_invokeThrows_publicRunThrows() throws InvocationTargetException, IllegalAccessException { thrown.expect(NullPointerException.class); doThrow(NullPointerException.class).when(invoker).invoke(any(InvocationContext.class)); sut.run(); }
@Test public void run_usingInterceptor_interceptorStartCalled() throws InvocationTargetException, IllegalAccessException { when(sagaInstanceInfo.isStarting()).thenReturn(true); sut.run(); InOrder inOrder = inOrder(invoker, interceptor); inOrder.verify(interceptor).onStarting(saga, context, theMessage); inOrder.verify(invoker).invoke(isA(InvocationContext.class)); }
@Test public void run_usingInterceptor_interceptorNotCalled() throws InvocationTargetException, IllegalAccessException { when(sagaInstanceInfo.isStarting()).thenReturn(false); sut.run(); verify(interceptor, never()).onStarting(any(Saga.class), any(ExecutionContext.class), any()); }
@Test public void run_usingInterceptorSagaFinished_interceptorFinishedCalled() throws InvocationTargetException, IllegalAccessException { when(sagaInstanceInfo.isStarting()).thenReturn(true); when(saga.isFinished()).thenReturn(true); sut.run(); InOrder inOrder = inOrder(invoker, interceptor); inOrder.verify(invoker).invoke(isA(InvocationContext.class)); inOrder.verify(interceptor).onFinished(saga, context); }
@Test public void run_usingInterceptorSagaNotFinished_interceptorFinishedNotCalled() throws InvocationTargetException, IllegalAccessException { when(sagaInstanceInfo.isStarting()).thenReturn(true); when(saga.isFinished()).thenReturn(false); sut.run(); verify(interceptor, never()).onFinished(any(Saga.class), any(ExecutionContext.class)); }
@Test public void run_usingInterceptor_interceptorHandlerExecutingCalled() throws InvocationTargetException, IllegalAccessException { when(sagaInstanceInfo.isStarting()).thenReturn(true); sut.run(); InOrder inOrder = inOrder(invoker, interceptor); inOrder.verify(interceptor).onHandlerExecuting(saga, context, theMessage); inOrder.verify(invoker).invoke(isA(InvocationContext.class)); }
@Test public void run_usingInterceptor_interceptorHandlerExecutedCalled() throws InvocationTargetException, IllegalAccessException { when(sagaInstanceInfo.isStarting()).thenReturn(true); sut.run(); InOrder inOrder = inOrder(invoker, interceptor); inOrder.verify(invoker).invoke(isA(InvocationContext.class)); inOrder.verify(interceptor).onHandlerExecuted(saga, context, theMessage); }
@Test public void run_sagaModuleStopsDispatching_invokeShallNotBeCalled() throws InvocationTargetException, IllegalAccessException { doAnswer(invocationOnMock -> { ((ExecutionContext) invocationOnMock.getArguments()[0]).stopDispatchingCurrentMessageToHandlers(); return null; }).when(module).onStart(any()); sut.run(); verify(invoker, never()).invoke(isA(InvocationContext.class)); }
@Test public void run_invokerAndModuleFinishThrow_throwsSagaExecutionExceptions() throws InvocationTargetException, IllegalAccessException { doThrow(ArithmeticException.class).when(invoker).invoke(isA(InvocationContext.class)); doThrow(NullPointerException.class).when(module).onFinished(any(ExecutionContext.class)); catchException(() -> sut.run()); SagaExecutionErrorsException exception = caughtException(); assertThat("Expected two exceptions in resulting one.", exception.getExecutionErrors(), hasSize(2)); assertThat("Expected two exceptions in resulting one.", exception.getExecutionErrors(), hasItem(instanceOf(NullPointerException.class))); assertThat("Expected two exceptions in resulting one.", exception.getExecutionErrors(), hasItem(instanceOf(ArithmeticException.class))); }