diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineEventResult.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineEventResult.java index added775..f5dce4d4 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineEventResult.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineEventResult.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2019-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,6 +18,8 @@ package org.springframework.statemachine; import org.springframework.messaging.Message; import org.springframework.statemachine.region.Region; +import reactor.core.publisher.Mono; + /** * Interface defining a result for sending an event to a statemachine. * @@ -49,6 +51,13 @@ public interface StateMachineEventResult { */ ResultType getResultType(); + /** + * Gets a mono representing completion. + * + * @return the mono for completion + */ + Mono complete(); + /** * Enumeration of a result type indicating whether a region accepted, denied or * deferred an event. @@ -60,17 +69,37 @@ public interface StateMachineEventResult { } /** - * Create a {@link StateMachineEventResult} from a {@link Region}, {@link Message} and a {@link ResultType}. + * Create a {@link StateMachineEventResult} from a {@link Region}, + * {@link Message} and a {@link ResultType}. * - * @param the type of state - * @param the type of event - * @param region the region - * @param message the message + * @param the type of state + * @param the type of event + * @param region the region + * @param message the message * @param resultType the result type * @return the state machine event result */ - public static StateMachineEventResult from(Region region, Message message, ResultType resultType) { - return new DefaultStateMachineEventResult<>(region, message, resultType); + public static StateMachineEventResult from(Region region, Message message, + ResultType resultType) { + return new DefaultStateMachineEventResult<>(region, message, resultType, null); + } + + + /** + * Create a {@link StateMachineEventResult} from a {@link Region}, + * {@link Message}, a {@link ResultType} and completion {@link Mono}. + * + * @param the type of state + * @param the type of event + * @param region the region + * @param message the message + * @param resultType the result type + * @param complete the completion mono + * @return the state machine event result + */ + public static StateMachineEventResult from(Region region, Message message, + ResultType resultType, Mono complete) { + return new DefaultStateMachineEventResult<>(region, message, resultType, complete); } static class DefaultStateMachineEventResult implements StateMachineEventResult { @@ -78,11 +107,14 @@ public interface StateMachineEventResult { private final Region region; private final Message message; private final ResultType resultType; + private Mono complete; - DefaultStateMachineEventResult(Region region, Message message, ResultType resultType) { + DefaultStateMachineEventResult(Region region, Message message, ResultType resultType, + Mono complete) { this.region = region; this.message = message; this.resultType = resultType; + this.complete = complete != null ? complete : Mono.empty(); } @Override @@ -100,6 +132,11 @@ public interface StateMachineEventResult { return resultType; } + @Override + public Mono complete() { + return complete; + } + @Override public String toString() { return "DefaultStateMachineEventResult [region=" + region + ", message=" + message + ", resultType=" diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineSystemConstants.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineSystemConstants.java index abfee972..bea472d5 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineSystemConstants.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineSystemConstants.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -34,4 +34,7 @@ public abstract class StateMachineSystemConstants { /** State machine id key for headers and variables */ public static final String STATEMACHINE_IDENTIFIER = "_sm_id_"; + + /** Contstant storing errors in a reactor context */ + public static final String REACTOR_CONTEXT_ERRORS = "stateMachineErrors"; } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ObjectState.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ObjectState.java index 2623289c..c890a8ea 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ObjectState.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ObjectState.java @@ -18,9 +18,12 @@ package org.springframework.statemachine.state; import java.util.Collection; import java.util.function.Function; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.region.Region; +import org.springframework.statemachine.support.StateMachineUtils; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -35,6 +38,8 @@ import reactor.core.publisher.Mono; */ public class ObjectState extends AbstractSimpleState { + private static final Log log = LogFactory.getLog(ObjectState.class); + /** * Instantiates a new object state. * @@ -145,17 +150,23 @@ public class ObjectState extends AbstractSimpleState { @Override public Mono exit(StateContext context) { Mono actions = Flux.fromIterable(getExitActions()) - .flatMap(a -> executeAction(a, context)) - .onErrorResume(t -> Mono.empty()) + .flatMap(a -> executeAction(a, context) + .doOnError(e -> { + log.warn("Exit action execution error", e); + })) + .onErrorResume(StateMachineUtils.resumeErrorToContext()) .then(); return super.exit(context).and(actions); } @Override public Mono entry(StateContext context) { - Mono actions = Flux.fromIterable(getEntryActions()) - .flatMap(a -> executeAction(a, context)) - .onErrorResume(t -> Mono.empty()) + Mono actions = Flux.fromIterable(getEntryActions()) + .flatMap(a -> executeAction(a, context) + .doOnError(e -> { + log.warn("Entry action execution error", e); + })) + .onErrorResume(StateMachineUtils.resumeErrorToContext()) .then(); return actions.and(super.entry(context)); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java index d710ce84..20d397f3 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java @@ -58,6 +58,7 @@ import org.springframework.statemachine.state.PseudoStateKind; import org.springframework.statemachine.state.PseudoStateListener; import org.springframework.statemachine.state.State; import org.springframework.statemachine.state.StateListenerAdapter; +import org.springframework.statemachine.support.StateMachineExecutor.MonoSinkStateMachineExecutorCallback; import org.springframework.statemachine.support.StateMachineExecutor.StateMachineExecutorTransit; import org.springframework.statemachine.transition.InitialTransition; import org.springframework.statemachine.transition.Transition; @@ -619,9 +620,8 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo } return Mono.just(message) .map(m -> getStateMachineInterceptors().preEvent(m, this)) - // .onErrorResume(error -> Mono.empty()) - .onErrorResume(error -> Mono.empty()) .flatMapMany(m -> acceptEvent(m)) + .onErrorResume(error -> Flux.just(StateMachineEventResult.from(this, message, ResultType.DENIED))) .doOnNext(notifyOnDenied()); } @@ -656,9 +656,11 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo return Mono.from(transition.getTrigger().evaluate(triggerContext)) .flatMap(e -> { if (e) { - return stateMachineExecutor.queueEvent(Mono.just(message)) + MonoSinkStateMachineExecutorCallback callback = new MonoSinkStateMachineExecutorCallback(); + Mono sink = Mono.create(callback); + return stateMachineExecutor.queueEvent(Mono.just(message), callback) .then(Mono.defer(() -> { - return Mono.just(StateMachineEventResult.from(this, message, ResultType.ACCEPTED)); + return Mono.just(StateMachineEventResult.from(this, message, ResultType.ACCEPTED, sink)); })) .onErrorResume(t -> { return Mono.defer(() -> { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java index 41b2c305..033b3fb6 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java @@ -22,6 +22,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Map.Entry; import java.util.Queue; import java.util.Set; @@ -55,6 +56,7 @@ import reactor.core.publisher.EmitterProcessor; import reactor.core.publisher.Flux; import reactor.core.publisher.FluxSink; import reactor.core.publisher.Mono; +import reactor.util.context.Context; /** * Default reactive implementation of a {@link StateMachineExecutor}. @@ -153,7 +155,7 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i if (log.isDebugEnabled()) { log.debug("Queue trigger " + trigger); } - triggerSink.next(new TriggerQueueItem(trigger, message)); + triggerSink.next(new TriggerQueueItem(trigger, message, null)); } @Override @@ -196,10 +198,10 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i } @Override - public Mono queueEvent(Mono> message) { + public Mono queueEvent(Mono> message, StateMachineExecutorCallback callback) { Flux> messages = Flux.merge(message, Flux.fromIterable(deferList)); return messages - .flatMap(m -> handleEvent(m)) + .flatMap(m -> handleEvent(m, callback)) .doOnNext(i -> { try { triggerSink.next(i); @@ -207,10 +209,11 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i throw new StateMachineException("Unable to handle queued event", e); } }) - .then(); + .then() + ; } - private Mono handleEvent(Message queuedEvent) { + private Mono handleEvent(Message queuedEvent, StateMachineExecutorCallback callback) { if (log.isDebugEnabled()) { log.debug("Handling message " + queuedEvent); } @@ -218,7 +221,7 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i State currentState = stateMachine.getState(); if ((currentState != null && currentState.shouldDefer(queuedEvent))) { log.info("Current state " + currentState + " deferred event " + queuedEvent); - return Mono.just(new TriggerQueueItem(null, queuedEvent)); + return Mono.just(new TriggerQueueItem(null, queuedEvent, callback)); } TriggerContext triggerContext = new DefaultTriggerContext(queuedEvent.getPayload()); return Flux.fromIterable(transitions) @@ -237,7 +240,7 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i }) .next() .doOnNext(trigger -> deferList.remove(queuedEvent)) - .map(trigger -> new TriggerQueueItem(trigger, queuedEvent)); + .map(trigger -> new TriggerQueueItem(trigger, queuedEvent, callback)); }); } @@ -303,7 +306,22 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i ret = Mono.empty(); } return ret; - }); + }) + .and(Mono.subscriberContext() + .doOnNext(ctx -> { + if (queueItem.callback != null) { + Optional holder = ctx.getOrEmpty(StateMachineSystemConstants.REACTOR_CONTEXT_ERRORS); + holder.ifPresent(h -> { + if (h.getError() != null) { + queueItem.callback.error(new StateMachineException("Execution error", h.getError())); + } else { + queueItem.callback.complete(); + } + }); + } + })) + .subscriberContext(Context.of(StateMachineSystemConstants.REACTOR_CONTEXT_ERRORS, new ExecutorExceptionHolder())) + ; } @@ -388,12 +406,7 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i .thenReturn(true) .doOnNext(a -> { interceptors.postTransition(stateContext); - }) - .onErrorResume(e -> { - interceptors.postTransition(stateContext); - return Mono.just(false); - }) - ; + }); } else { return Mono.just(false); } @@ -457,9 +470,12 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i private class TriggerQueueItem { Trigger trigger; Message message; - public TriggerQueueItem(Trigger trigger, Message message) { + StateMachineExecutorCallback callback; + + public TriggerQueueItem(Trigger trigger, Message message, StateMachineExecutorCallback callback) { this.trigger = trigger; this.message = message; + this.callback = callback; } } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java index edc35bed..91b6a8f0 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java @@ -15,6 +15,8 @@ */ package org.springframework.statemachine.support; +import java.util.function.Consumer; + import org.springframework.messaging.Message; import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; @@ -24,6 +26,7 @@ import org.springframework.statemachine.transition.Transition; import org.springframework.statemachine.trigger.Trigger; import reactor.core.publisher.Mono; +import reactor.core.publisher.MonoSink; /** * Interface for a {@link StateMachine} event executor. @@ -39,9 +42,10 @@ public interface StateMachineExecutor extends StateMachineReactiveLifecycl * Queue event. * * @param message the message + * @param callback the executor callback * @return completion when event is queued */ - Mono queueEvent(Mono> message); + Mono queueEvent(Mono> message, StateMachineExecutorCallback callback); /** * Queue trigger. @@ -111,4 +115,52 @@ public interface StateMachineExecutor extends StateMachineReactiveLifecycl */ Mono transit(Transition transition, StateContext stateContext, Message message); } + + /** + * Completion callback to notify back complete or error. + */ + public interface StateMachineExecutorCallback { + void complete(); + void error(Throwable e); + } + + static class MonoSinkStateMachineExecutorCallback implements Consumer>, StateMachineExecutorCallback { + + private boolean complete; + private Throwable error; + + @Override + public void complete() { + complete = true; + } + + @Override + public void error(Throwable e) { + error = e; + } + + @Override + public void accept(MonoSink t) { + if (complete) { + t.success(); + } else if (error != null) { + t.error(error); + } else { + t.success(); + } + } + } + + public static class ExecutorExceptionHolder { + + private Throwable error; + + public void setError(Throwable error) { + this.error = error; + } + + public Throwable getError() { + return error; + } + } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineUtils.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineUtils.java index a3d15ee9..337e48e8 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineUtils.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,14 +17,19 @@ package org.springframework.statemachine.support; import java.util.ArrayList; import java.util.Collection; +import java.util.Optional; import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachineMessageHeaders; +import org.springframework.statemachine.StateMachineSystemConstants; import org.springframework.statemachine.state.PseudoState; import org.springframework.statemachine.state.PseudoStateKind; import org.springframework.statemachine.state.State; +import org.springframework.statemachine.support.StateMachineExecutor.ExecutorExceptionHolder; import org.springframework.util.ObjectUtils; +import reactor.core.publisher.Mono; + /** * Various utility methods for state machine. * @@ -191,4 +196,20 @@ public abstract class StateMachineUtils { } return null; } + + /** + * Utility function to stash error into reactor context. + * + * @return mono for completion + */ + public static java.util.function.Function> resumeErrorToContext() { + return t -> Mono.subscriberContext() + .doOnNext(ctx -> { + Optional holder = ctx.getOrEmpty(StateMachineSystemConstants.REACTOR_CONTEXT_ERRORS); + holder.ifPresent(h -> { + h.setError(t); + }); + }) + .then(); + } } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/StateMachineErrorTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/StateMachineErrorTests.java index 56e60071..4919f29a 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/StateMachineErrorTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/StateMachineErrorTests.java @@ -26,6 +26,7 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; import org.springframework.statemachine.config.EnableStateMachine; import org.springframework.statemachine.config.EnumStateMachineConfigurerAdapter; import org.springframework.statemachine.config.StateMachineConfigurerAdapter; @@ -39,6 +40,9 @@ import org.springframework.statemachine.state.State; import org.springframework.statemachine.support.StateMachineInterceptorAdapter; import org.springframework.statemachine.transition.Transition; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + /** * Tests for various errors and error handling. * @@ -507,4 +511,50 @@ public class StateMachineErrorTests extends AbstractStateMachineTests { .target("S2"); } } + + @SuppressWarnings("unchecked") + @Test + public void testActionEntryErrorWithEvent() throws Exception { + context.register(Config4.class); + context.refresh(); + + ObjectStateMachine machine = + context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + machine.start(); + assertThat(machine.getState().getIds()).containsExactlyInAnyOrder("SI"); + + StepVerifier.create(machine.sendEvent(Mono.just(MessageBuilder.withPayload("E1").build()))) + .consumeNextWith(result -> { + StepVerifier.create(result.complete()).consumeErrorWith(e -> { + assertThat(e).isInstanceOf(StateMachineException.class).hasMessageContaining("Execution error"); + }).verify(); + }) + .verifyComplete(); + + assertThat(machine.getState().getIds()).containsExactlyInAnyOrder("S1"); + } + + @Configuration + @EnableStateMachine + static class Config4 extends StateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial("SI") + .stateEntry("S1", (context) -> { + throw new RuntimeException("error"); + }); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions + .withExternal() + .source("SI") + .target("S1") + .event("E1"); + } + } } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/security/EventSecurityTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/security/EventSecurityTests.java index 55a1022f..def627da 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/security/EventSecurityTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/security/EventSecurityTests.java @@ -43,7 +43,7 @@ public class EventSecurityTests extends AbstractSecurityTests { public void testEventDeniedViaExpression() throws Exception { TestListener listener = new TestListener(); StateMachine machine = buildMachine(listener, null, null, null, null, null, "false"); - assertTransitionDenied(machine, listener); + assertTransitionDeniedResultAsDenied(machine, listener); } @Test @@ -51,7 +51,7 @@ public class EventSecurityTests extends AbstractSecurityTests { public void testEventDeniedViaAttributes() throws Exception { TestListener listener = new TestListener(); StateMachine machine = buildMachine(listener, null, null, null, "EVENT_B", ComparisonType.ALL, null); - assertTransitionDenied(machine, listener); + assertTransitionDeniedResultAsDenied(machine, listener); } @Test