Pass error from executor

- New concept of using callback to pass on error as
  things can't be properly propagated as we're manually
  dispatching to processor.
- Relates #821

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<S, E> {
 	 */
 	ResultType getResultType();
 
+	/**
+	 * Gets a mono representing completion.
+	 *
+	 * @return the mono for completion
+	 */
+	Mono<Void> complete();
+
 	/**
 	 * Enumeration of a result type indicating whether a region accepted, denied or
 	 * deferred an event.
@@ -60,17 +69,37 @@ public interface StateMachineEventResult<S, E> {
 	}
 
 	/**
-	 * 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 <S> the type of state
-	 * @param <E> the type of event
-	 * @param region the region
-	 * @param message the message
+	 * @param <S>        the type of state
+	 * @param <E>        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 <S, E> StateMachineEventResult<S, E> from(Region<S, E> region, Message<E> message, ResultType resultType) {
-		return new DefaultStateMachineEventResult<>(region, message, resultType);
+	public static <S, E> StateMachineEventResult<S, E> from(Region<S, E> region, Message<E> 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 <S>        the type of state
+	 * @param <E>        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 <S, E> StateMachineEventResult<S, E> from(Region<S, E> region, Message<E> message,
+			ResultType resultType, Mono<Void> complete) {
+		return new DefaultStateMachineEventResult<>(region, message, resultType, complete);
 	}
 
 	static class DefaultStateMachineEventResult<S, E> implements StateMachineEventResult<S, E> {
@@ -78,11 +107,14 @@ public interface StateMachineEventResult<S, E> {
 		private final Region<S, E> region;
 		private final Message<E> message;
 		private final ResultType resultType;
+		private Mono<Void> complete;
 
-		DefaultStateMachineEventResult(Region<S, E> region, Message<E> message, ResultType resultType) {
+		DefaultStateMachineEventResult(Region<S, E> region, Message<E> message, ResultType resultType,
+				Mono<Void> 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<S, E> {
 			return resultType;
 		}
 
+		@Override
+		public Mono<Void> complete() {
+			return complete;
+		}
+
 		@Override
 		public String toString() {
 			return "DefaultStateMachineEventResult [region=" + region + ", message=" + message + ", resultType="

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";
 }

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<S, E> extends AbstractSimpleState<S, E> {
 
+	private static final Log log = LogFactory.getLog(ObjectState.class);
+
 	/**
 	 * Instantiates a new object state.
 	 *
@@ -145,17 +150,23 @@ public class ObjectState<S, E> extends AbstractSimpleState<S, E> {
 	@Override
 	public Mono<Void> exit(StateContext<S, E> context) {
 		Mono<Void> 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<Void> entry(StateContext<S, E> context) {
-		Mono<Void> actions = Flux.fromIterable(getEntryActions())
-			.flatMap(a -> executeAction(a, context))
-			.onErrorResume(t -> Mono.empty())
+		Mono<Void> 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));
 	}

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<S, E> 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.<S, E>from(this, message, ResultType.DENIED)))
 			.doOnNext(notifyOnDenied());
 	}
 
@@ -656,9 +656,11 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
 									return Mono.from(transition.getTrigger().evaluate(triggerContext))
 										.flatMap(e -> {
 											if (e) {
-												return stateMachineExecutor.queueEvent(Mono.just(message))
+												MonoSinkStateMachineExecutorCallback callback = new MonoSinkStateMachineExecutorCallback();
+												Mono<Void> sink = Mono.create(callback);
+												return stateMachineExecutor.queueEvent(Mono.just(message), callback)
 													.then(Mono.defer(() -> {
-														return Mono.just(StateMachineEventResult.<S, E>from(this, message, ResultType.ACCEPTED));
+														return Mono.just(StateMachineEventResult.<S, E>from(this, message, ResultType.ACCEPTED, sink));
 													}))
 													.onErrorResume(t -> {
 														return Mono.defer(() -> {

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<S, E> 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<S, E> extends LifecycleObjectSupport i
 	}
 
 	@Override
-	public Mono<Void> queueEvent(Mono<Message<E>> message) {
+	public Mono<Void> queueEvent(Mono<Message<E>> message, StateMachineExecutorCallback callback) {
 		Flux<Message<E>> 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<S, E> extends LifecycleObjectSupport i
 					throw new StateMachineException("Unable to handle queued event", e);
 				}
 			})
-			.then();
+			.then()
+			;
 	}
 
-	private Mono<TriggerQueueItem> handleEvent(Message<E> queuedEvent) {
+	private Mono<TriggerQueueItem> handleEvent(Message<E> queuedEvent, StateMachineExecutorCallback callback) {
 		if (log.isDebugEnabled()) {
 			log.debug("Handling message " + queuedEvent);
 		}
@@ -218,7 +221,7 @@ public class ReactiveStateMachineExecutor<S, E> extends LifecycleObjectSupport i
 			State<S,E> 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<S, E> triggerContext = new DefaultTriggerContext<S, E>(queuedEvent.getPayload());
 			return Flux.fromIterable(transitions)
@@ -237,7 +240,7 @@ public class ReactiveStateMachineExecutor<S, E> 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<S, E> extends LifecycleObjectSupport i
 				ret = Mono.empty();
 			}
 			return ret;
-		});
+		})
+		.and(Mono.subscriberContext()
+			.doOnNext(ctx -> {
+				if (queueItem.callback != null) {
+					Optional<ExecutorExceptionHolder> 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<S, E> 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<S, E> extends LifecycleObjectSupport i
 	private class TriggerQueueItem {
 		Trigger<S, E> trigger;
 		Message<E> message;
-		public TriggerQueueItem(Trigger<S, E> trigger, Message<E> message) {
+		StateMachineExecutorCallback callback;
+
+		public TriggerQueueItem(Trigger<S, E> trigger, Message<E> message, StateMachineExecutorCallback callback) {
 			this.trigger = trigger;
 			this.message = message;
+			this.callback = callback;
 		}
 	}
 }

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<S, E> extends StateMachineReactiveLifecycl
 	 * Queue event.
 	 *
 	 * @param message the message
+	 * @param callback the executor callback
 	 * @return completion when event is queued
 	 */
-	Mono<Void> queueEvent(Mono<Message<E>> message);
+	Mono<Void> queueEvent(Mono<Message<E>> message, StateMachineExecutorCallback callback);
 
 	/**
 	 * Queue trigger.
@@ -111,4 +115,52 @@ public interface StateMachineExecutor<S, E> extends StateMachineReactiveLifecycl
 		 */
 		Mono<Void> transit(Transition<S, E> transition, StateContext<S, E> stateContext, Message<E> message);
 	}
+
+	/**
+	 * Completion callback to notify back complete or error.
+	 */
+	public interface StateMachineExecutorCallback {
+		void complete();
+		void error(Throwable e);
+	}
+
+	static class MonoSinkStateMachineExecutorCallback implements Consumer<MonoSink<Void>>, 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<Void> 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;
+		}
+	}
 }

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<? super Throwable, Mono<Void>> resumeErrorToContext() {
+		return t -> Mono.subscriberContext()
+			.doOnNext(ctx -> {
+				Optional<ExecutorExceptionHolder> holder = ctx.getOrEmpty(StateMachineSystemConstants.REACTOR_CONTEXT_ERRORS);
+				holder.ifPresent(h -> {
+					h.setError(t);
+				});
+			})
+			.then();
+	}
 }

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<String, String> 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<String, String> {
+
+		@Override
+		public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
+			states
+				.withStates()
+					.initial("SI")
+					.stateEntry("S1", (context) -> {
+						throw new RuntimeException("error");
+					});
+		}
+
+		@Override
+		public void configure(StateMachineTransitionConfigurer<String, String> transitions) throws Exception {
+			transitions
+				.withExternal()
+					.source("SI")
+					.target("S1")
+					.event("E1");
+		}
+	}
 }

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<States, Events> 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<States, Events> machine = buildMachine(listener, null, null, null, "EVENT_B", ComparisonType.ALL, null);
-		assertTransitionDenied(machine, listener);
+		assertTransitionDeniedResultAsDenied(machine, listener);
 	}
 
 	@Test