Reactive PseudoStates

- Change entry/exit functions to reactive types
  to have those in a reactive stack.
- Fixes #878

spring-statemachine-build-tests/src/test/java/org/springframework/statemachine/buildtests/ForkJoinEntryExitTests.java

@@ -92,6 +92,7 @@ public class ForkJoinEntryExitTests extends AbstractBuildTests {
 					.step()
 						.sendEvent("E3")
 						.expectStateEntered(2)
+						// TODO: S211 exited twice
 						.expectStateExited(4)
 						.expectStates("S3").and()
 					.build();
@@ -122,6 +123,7 @@ public class ForkJoinEntryExitTests extends AbstractBuildTests {
 					.step()
 						.sendEvent("E3")
 						.expectStateEntered(2)
+						// TODO: S211 exited twice
 						.expectStateExited(4)
 						.expectStates("S3").and()
 					.build();

spring-statemachine-core/src/main/java/org/springframework/statemachine/action/Actions.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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.
@@ -16,7 +16,10 @@
 
 package org.springframework.statemachine.action;
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import org.springframework.statemachine.StateContext;
 import org.springframework.statemachine.support.DefaultStateContext;
@@ -97,4 +100,21 @@ public final class Actions {
 			return null;
 		}
 	}
+
+
+	/**
+	 * Builds a {@link Collection} of {@link Function}s from a {@link Collection} of an {@link Action}s.
+	 *
+	 * @param <S> the type of state
+	 * @param <E> the type of event
+	 * @param actions the actions
+	 * @return the function
+	 */
+	public static <S, E> Collection<Function<StateContext<S, E>, Mono<Void>>> from(Collection<Action<S, E>> actions) {
+		if (actions != null) {
+			return actions.stream().map(action -> from(action)).collect(Collectors.toList());
+		} else {
+			return Collections.emptyList();
+		}
+	}
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java

@@ -722,7 +722,7 @@ public abstract class AbstractStateMachineFactory<S, E> extends LifecycleObjectS
 					if (holder.getState() == null) {
 						holderList.add(new HolderListItem<S, E>(c.getTarget(), holder));
 					}
-					choices.add(new ChoiceStateData<S, E>(holder, c.getGuard(), c.getActions()));
+					choices.add(new ChoiceStateData<S, E>(holder, c.getGuard(), Actions.from(c.getActions())));
 				}
 				PseudoState<S, E> pseudoState = new ChoicePseudoState<S, E>(choices);
 				state = buildStateInternal(stateData.getState(), stateData.getDeferred(), stateData.getEntryActions(),
@@ -741,7 +741,7 @@ public abstract class AbstractStateMachineFactory<S, E> extends LifecycleObjectS
 					if (holder.getState() == null) {
 						holderList.add(new HolderListItem<S, E>(c.getTarget(), holder));
 					}
-					junctions.add(new JunctionStateData<S, E>(holder, c.getGuard(), c.getActions()));
+					junctions.add(new JunctionStateData<S, E>(holder, c.getGuard(), Actions.from(c.getActions())));
 				}
 				PseudoState<S, E> pseudoState = new JunctionPseudoState<S, E>(junctions);
 				state = buildStateInternal(stateData.getState(), stateData.getDeferred(), stateData.getEntryActions(),

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/AbstractPseudoState.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.
@@ -19,6 +19,8 @@ import java.util.List;
 
 import org.springframework.statemachine.StateContext;
 
+import reactor.core.publisher.Mono;
+
 /**
  * Base implementation of a {@link PseudoState}.
  *
@@ -48,12 +50,13 @@ public abstract class AbstractPseudoState<S, E> implements PseudoState<S, E> {
 	}
 
 	@Override
-	public State<S, E> entry(StateContext<S, E> context) {
-		return null;
+	public Mono<State<S, E>> entry(StateContext<S, E> context) {
+		return Mono.empty();
 	}
 
 	@Override
-	public void exit(StateContext<S, E> context) {
+	public Mono<Void> exit(StateContext<S, E> context) {
+		return Mono.empty();
 	}
 
 	@Override

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ChoicePseudoState.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-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,17 @@ package org.springframework.statemachine.state;
 
 import java.util.Collection;
 import java.util.List;
+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.action.Action;
 import org.springframework.statemachine.guard.Guard;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 /**
  * Choice implementation of a {@link PseudoState}.
  *
@@ -53,24 +56,27 @@ public class ChoicePseudoState<S, E> implements PseudoState<S, E> {
 	}
 
 	@Override
-	public State<S, E> entry(StateContext<S, E> context) {
-		State<S, E> s = null;
-		ChoiceStateData<S, E> csd = null;
-		for (ChoiceStateData<S, E> c : choices) {
-			csd = c;
-			if (c.guard != null && evaluateInternal(c.guard, context)) {
-				break;
+	public Mono<State<S, E>> entry(StateContext<S, E> context) {
+		return Mono.defer(() -> {
+			ChoiceStateData<S, E> csd = null;
+			for (ChoiceStateData<S, E> c : choices) {
+				csd = c;
+				if (c.guard != null && evaluateInternal(c.guard, context)) {
+					break;
+				}
 			}
-		}
-		if (csd != null) {
-			s = csd.getState();
-			executeActions(csd.getActions(), context);
-		}
-		return s;
+			return Mono.justOrEmpty(csd);
+		})
+		.flatMap(csd -> {
+			return Flux.fromIterable(csd.getActions())
+				.flatMap(a -> a.apply(context))
+				.then(Mono.just(csd.getState()));
+		});
 	}
 
 	@Override
-	public void exit(StateContext<S, E> context) {
+	public Mono<Void> exit(StateContext<S, E> context) {
+		return Mono.empty();
 	}
 
 	@Override
@@ -90,19 +96,6 @@ public class ChoicePseudoState<S, E> implements PseudoState<S, E> {
 		}
 	}
 
-	private void executeActions(Collection<Action<S, E>> actions, StateContext<S, E> context) {
-		if (actions == null) {
-			return;
-		}
-		for (Action<S, E> action : actions) {
-			try {
-				action.execute(context);
-			} catch (Throwable t) {
-				log.warn("Action execution resulted error", t);
-			}
-		}
-	}
-
 	/**
 	 * Data class wrapping choice {@link State} and {@link Guard}
 	 * together.
@@ -113,7 +106,7 @@ public class ChoicePseudoState<S, E> implements PseudoState<S, E> {
 	public static class ChoiceStateData<S, E> {
 		private final StateHolder<S, E> state;
 		private final Guard<S, E> guard;
-		private final Collection<Action<S, E>> actions;
+		private final Collection<Function<StateContext<S, E>, Mono<Void>>> actions;
 
 		/**
 		 * Instantiates a new choice state data.
@@ -122,7 +115,8 @@ public class ChoicePseudoState<S, E> implements PseudoState<S, E> {
 		 * @param guard the guard
 		 * @param actions the actions
 		 */
-		public ChoiceStateData(StateHolder<S, E> state, Guard<S, E> guard, Collection<Action<S, E>> actions) {
+		public ChoiceStateData(StateHolder<S, E> state, Guard<S, E> guard,
+				Collection<Function<StateContext<S, E>, Mono<Void>>> actions) {
 			Assert.notNull(state, "Holder must be set");
 			this.state = state;
 			this.guard = guard;
@@ -161,7 +155,7 @@ public class ChoicePseudoState<S, E> implements PseudoState<S, E> {
 		 *
 		 * @return the actions
 		 */
-		public Collection<Action<S, E>> getActions() {
+		public Collection<Function<StateContext<S, E>, Mono<Void>>> getActions() {
 			return actions;
 		}
 	}

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/EntryPseudoState.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-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.
@@ -19,6 +19,8 @@ import java.util.List;
 
 import org.springframework.statemachine.StateContext;
 
+import reactor.core.publisher.Mono;
+
 /**
  * Entrypoint implementation of a {@link PseudoState}.
  *
@@ -46,12 +48,13 @@ public class EntryPseudoState<S, E> implements PseudoState<S, E> {
 	}
 
 	@Override
-	public State<S, E> entry(StateContext<S, E> context) {
-		return state;
+	public Mono<State<S, E>> entry(StateContext<S, E> context) {
+		return Mono.just(state);
 	}
 
 	@Override
-	public void exit(StateContext<S, E> context) {
+	public Mono<Void> exit(StateContext<S, E> context) {
+		return Mono.empty();
 	}
 
 	@Override

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ExitPseudoState.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-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.
@@ -20,6 +20,8 @@ import java.util.List;
 import org.springframework.statemachine.StateContext;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Mono;
+
 /**
  * Exitpoint implementation of a {@link PseudoState}.
  *
@@ -48,12 +50,13 @@ public class ExitPseudoState<S, E> implements PseudoState<S, E> {
 	}
 
 	@Override
-	public State<S, E> entry(StateContext<S, E> context) {
-		return state.getState();
+	public Mono<State<S, E>> entry(StateContext<S, E> context) {
+		return Mono.just(state.getState());
 	}
 
 	@Override
-	public void exit(StateContext<S, E> context) {
+	public Mono<Void> exit(StateContext<S, E> context) {
+		return Mono.empty();
 	}
 
 	@Override

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ForkPseudoState.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.
@@ -19,6 +19,8 @@ import java.util.List;
 
 import org.springframework.statemachine.StateContext;
 
+import reactor.core.publisher.Mono;
+
 /**
  * Fork implementation of a {@link PseudoState}.
  *
@@ -37,12 +39,11 @@ public class ForkPseudoState<S, E> extends AbstractPseudoState<S, E> {
 	}
 
 	@Override
-	public State<S, E> entry(StateContext<S, E> context) {
-		return null;
+	public Mono<State<S, E>> entry(StateContext<S, E> context) {
+		return Mono.empty();
 	}
 
 	public List<State<S, E>> getForks() {
 		return forks;
 	}
-
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/HistoryPseudoState.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.
@@ -18,6 +18,8 @@ package org.springframework.statemachine.state;
 import org.springframework.statemachine.StateContext;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Mono;
+
 /**
  * History implementation of a {@link PseudoState}.
  *
@@ -60,21 +62,21 @@ public class HistoryPseudoState<S, E> extends AbstractPseudoState<S, E> {
 	}
 
 	@Override
-	public State<S, E> entry(StateContext<S, E> context) {
+	public Mono<State<S, E>> entry(StateContext<S, E> context) {
 		// if no logged history or history is final state,
 		// go to default state. go to containing parent if
 		// we have no history and there's no default state.
 		if (state == null) {
 			if (defaultState.getState() == null) {
-				return containingState.getState();
+				return Mono.just(containingState.getState());
 			} else {
-				return defaultState.getState();
+				return Mono.just(defaultState.getState());
 			}
 		} else {
 			if (defaultState.getState() != null && state.getPseudoState() != null && state.getPseudoState().getKind() == PseudoStateKind.END) {
-				return defaultState.getState();
+				return Mono.just(defaultState.getState());
 			} else {
-				return state;
+				return Mono.just(state);
 			}
 		}
 	}

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/JoinPseudoState.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.
@@ -28,6 +28,7 @@ import org.springframework.statemachine.guard.Guard;
 import org.springframework.statemachine.state.PseudoStateContext.PseudoAction;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 /**
@@ -59,23 +60,23 @@ public class JoinPseudoState<S, E> extends AbstractPseudoState<S, E> {
 	}
 
 	@Override
-	public State<S, E> entry(StateContext<S, E> context) {
-		if (!tracker.isNotified()) {
-			return null;
-		}
-		State<S, E> s = null;
-		for (JoinStateData<S, E> c : joinTargets) {
-			s = c.getState();
-			if (c.guard != null && evaluateInternal(c.guard, context)) {
-				break;
+	public Mono<State<S, E>> entry(StateContext<S, E> context) {
+		return Mono.defer(() -> {
+			if (!tracker.isNotified()) {
+				return Mono.empty();
 			}
-		}
-		return s;
+			return Flux.fromIterable(joinTargets)
+				.filterWhen(jst -> evaluateInternal(jst.guard, context))
+				.next()
+				.map(jst -> jst.getState());
+		});
 	}
 
 	@Override
-	public void exit(StateContext<S, E> context) {
-		tracker.reset();
+	public Mono<Void> exit(StateContext<S, E> context) {
+		return Mono.fromRunnable(() -> {
+			tracker.reset();
+		});
 	}
 
 	/**
@@ -97,16 +98,15 @@ public class JoinPseudoState<S, E> extends AbstractPseudoState<S, E> {
 		tracker.reset(ids);
 	}
 
-	private boolean evaluateInternal(Function<StateContext<S, E>, Mono<Boolean>> guard, StateContext<S, E> context) {
+	private Mono<Boolean> evaluateInternal(Function<StateContext<S, E>, Mono<Boolean>> guard, StateContext<S, E> context) {
+		if (guard == null) {
+			return Mono.just(true);
+		}
 		try {
-			// Function<StateContext<S, E>, Mono<Boolean>>
-			// TODO: REACTOR no blocking!
-			// return guard.evaluate(context);
-			return guard.apply(context).block();
-
-		} catch (Throwable t) {
-			log.warn("Deny guard due to throw as GUARD should not error", t);
-			return false;
+			return guard.apply(context);
+		} catch (Exception e) {
+			log.warn("Deny guard due to throw as GUARD should not error");
+			return Mono.just(false);
 		}
 	}
 

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/JunctionPseudoState.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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,17 @@ package org.springframework.statemachine.state;
 
 import java.util.Collection;
 import java.util.List;
+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.action.Action;
 import org.springframework.statemachine.guard.Guard;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 /**
  * Junction implementation of a {@link PseudoState}.
  *
@@ -53,24 +56,28 @@ public class JunctionPseudoState<S, E> implements PseudoState<S, E> {
 	}
 
 	@Override
-	public State<S, E> entry(StateContext<S, E> context) {
-		State<S, E> s = null;
-		JunctionStateData<S, E> jsd = null;
-		for (JunctionStateData<S, E> j : junctions) {
-			jsd = j;
-			if (j.guard != null && evaluateInternal(j.guard, context)) {
-				break;
+	public Mono<State<S, E>> entry(StateContext<S, E> context) {
+		return Mono.defer(() -> {
+			JunctionStateData<S, E> jsd = null;
+			for (JunctionStateData<S, E> j : junctions) {
+				jsd = j;
+				if (j.guard != null && evaluateInternal(j.guard, context)) {
+					break;
+				}
 			}
-		}
-		if (jsd != null) {
-			s = jsd.getState();
-			executeActions(jsd.getActions(), context);
-		}
-		return s;
+			return Mono.justOrEmpty(jsd);
+		})
+		.flatMap(jsd -> {
+			return Flux.fromIterable(jsd.getActions())
+				.flatMap(a -> a.apply(context))
+				.then(Mono.just(jsd.getState()));
+		});
 	}
 
+
 	@Override
-	public void exit(StateContext<S, E> context) {
+	public Mono<Void> exit(StateContext<S, E> context) {
+		return Mono.empty();
 	}
 
 	@Override
@@ -90,19 +97,6 @@ public class JunctionPseudoState<S, E> implements PseudoState<S, E> {
 		}
 	}
 
-	private void executeActions(Collection<Action<S, E>> actions, StateContext<S, E> context) {
-		if (actions == null) {
-			return;
-		}
-		for (Action<S, E> action : actions) {
-			try {
-				action.execute(context);
-			} catch (Throwable t) {
-				log.warn("Action execution resulted error", t);
-			}
-		}
-	}
-
 	/**
 	 * Data class wrapping choice {@link State} and {@link Guard}
 	 * together.
@@ -113,7 +107,7 @@ public class JunctionPseudoState<S, E> implements PseudoState<S, E> {
 	public static class JunctionStateData<S, E> {
 		private final StateHolder<S, E> state;
 		private final Guard<S, E> guard;
-		private final Collection<Action<S, E>> actions;
+		private final Collection<Function<StateContext<S, E>, Mono<Void>>> actions;
 
 		/**
 		 * Instantiates a new junction state data.
@@ -122,7 +116,8 @@ public class JunctionPseudoState<S, E> implements PseudoState<S, E> {
 		 * @param guard the guard
 		 * @param actions the actions
 		 */
-		public JunctionStateData(StateHolder<S, E> state, Guard<S, E> guard, Collection<Action<S, E>> actions) {
+		public JunctionStateData(StateHolder<S, E> state, Guard<S, E> guard,
+				Collection<Function<StateContext<S, E>, Mono<Void>>> actions) {
 			Assert.notNull(state, "Holder must be set");
 			this.state = state;
 			this.guard = guard;
@@ -161,7 +156,7 @@ public class JunctionPseudoState<S, E> implements PseudoState<S, E> {
 		 *
 		 * @return the actions
 		 */
-		public Collection<Action<S, E>> getActions() {
+		public Collection<Function<StateContext<S, E>, Mono<Void>>> getActions() {
 			return actions;
 		}
 	}

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/PseudoState.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.
@@ -19,6 +19,8 @@ import java.util.List;
 
 import org.springframework.statemachine.StateContext;
 
+import reactor.core.publisher.Mono;
+
 /**
  * A {@code PseudoState} is an abstraction that encompasses different types of
  * transient states or vertices in the state machine.
@@ -51,14 +53,15 @@ public interface PseudoState<S, E> {
 	 * @param context the context
 	 * @return the next state or null
 	 */
-	State<S, E> entry(StateContext<S, E> context);
+	Mono<State<S, E>> entry(StateContext<S, E> context);
 
 	/**
 	 * Initiate an exit sequence for the state.
 	 *
 	 * @param context the context
+	 * @return mono of completion
 	 */
-	void exit(StateContext<S, E> context);
+	Mono<Void> exit(StateContext<S, E> context);
 
 	/**
 	 * Registers a new {@link PseudoStateListener}.

spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java

@@ -932,57 +932,55 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
 					&& !callPreStateChangeInterceptors(state, message, transition, stateMachine)) {
 				return Mono.empty();
 			}
-
 			StateContext<S, E> stateContext = buildStateContext(Stage.STATE_CHANGED, message, transition, stateMachine);
-			State<S,E> toState = followLinkedPseudoStates(state, stateContext);
-			PseudoStateKind kind = state.getPseudoState() != null ? state.getPseudoState().getKind() : null;
+			return Mono.from(followLinkedPseudoStates(state, stateContext))
+				.flatMap(toState -> {
+					PseudoStateKind kind = state.getPseudoState() != null ? state.getPseudoState().getKind() : null;
 
-			if (kind != null && (kind != PseudoStateKind.INITIAL && kind != PseudoStateKind.JOIN
-					&& kind != PseudoStateKind.FORK && kind != PseudoStateKind.END)) {
-				callPreStateChangeInterceptors(toState, message, transition, stateMachine);
-			}
+					if (kind != null && (kind != PseudoStateKind.INITIAL && kind != PseudoStateKind.JOIN
+							&& kind != PseudoStateKind.FORK && kind != PseudoStateKind.END)) {
+						callPreStateChangeInterceptors(toState, message, transition, stateMachine);
+					}
 
-			kind = toState.getPseudoState() != null ? toState.getPseudoState().getKind() : null;
-			if (kind == PseudoStateKind.FORK) {
-				Mono<Void> ret1 = exitCurrentState(toState, message, transition, stateMachine);
-				ForkPseudoState<S, E> fps = (ForkPseudoState<S, E>) toState.getPseudoState();
-				Mono<Void> ret2 = Flux.fromIterable(fps.getForks())
-					.flatMap(f -> {
-						callPreStateChangeInterceptors(f, message, transition, stateMachine);
-						return setCurrentState(f, message, transition, false, stateMachine, null, fps.getForks());
-					})
-					.then()
-					;
-				return ret1.then(ret2);
-			} else {
-				Collection<State<S, E>> targets = new ArrayList<>();
-				targets.add(toState);
-				return setCurrentState(toState, message, transition, true, stateMachine, null, targets);
-			}
+					kind = toState.getPseudoState() != null ? toState.getPseudoState().getKind() : null;
+					if (kind == PseudoStateKind.FORK) {
+						Mono<Void> ret1 = exitCurrentState(toState, message, transition, stateMachine);
+						ForkPseudoState<S, E> fps = (ForkPseudoState<S, E>) toState.getPseudoState();
+						Mono<Void> ret2 = Flux.fromIterable(fps.getForks())
+							.flatMap(f -> {
+								callPreStateChangeInterceptors(f, message, transition, stateMachine);
+								return setCurrentState(f, message, transition, false, stateMachine, null, fps.getForks());
+							})
+							.then()
+							;
+						return ret1.then(ret2);
+					} else {
+						Collection<State<S, E>> targets = new ArrayList<>();
+						targets.add(toState);
+						return setCurrentState(toState, message, transition, true, stateMachine, null, targets);
+					}
+
+				});
 		})
 		.then(Mono.defer(() -> {
 			return shouldComplete() ? stopReactively() : Mono.empty();
-		}))
-		;
+		}));
 	}
 
 	private boolean shouldComplete() {
 		return StateMachineUtils.isPseudoState(currentState, PseudoStateKind.END);
 	}
 
-	private State<S,E> followLinkedPseudoStates(State<S,E> state, StateContext<S, E> stateContext) {
+	private Mono<State<S,E>> followLinkedPseudoStates(State<S,E> state, StateContext<S, E> stateContext) {
 		PseudoStateKind kind = state.getPseudoState() != null ? state.getPseudoState().getKind() : null;
 		if (kind == PseudoStateKind.INITIAL ||  kind == PseudoStateKind.FORK) {
-			return state;
+			return Mono.just(state);
 		} else if (kind != null) {
-			State<S,E> toState = state.getPseudoState().entry(stateContext);
-			if (toState == null) {
-				return state;
-			} else {
-				return followLinkedPseudoStates(toState, stateContext);
-			}
+			return Mono.from(state.getPseudoState().entry(stateContext).log("xxx1").flatMap(s -> followLinkedPseudoStates(s, stateContext)))
+				.switchIfEmpty(Mono.just(state))
+				;
 		} else {
-			return state;
+			return Mono.just(state);
 		}
 	}
 
@@ -992,17 +990,26 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
 			if (p != null) {
 				List<PseudoStateListener<S, E>> listeners = new ArrayList<PseudoStateListener<S, E>>();
 				listeners.add(new PseudoStateListener<S, E>() {
+
 					@Override
 					public void onContext(PseudoStateContext<S, E> context) {
 						PseudoState<S, E> pseudoState = context.getPseudoState();
 						State<S, E> toStateOrig = findStateWithPseudoState(pseudoState);
 						StateContext<S, E> stateContext = buildStateContext(Stage.STATE_EXIT, null, null, getRelayStateMachine());
-						State<S, E> toState = followLinkedPseudoStates(toStateOrig, stateContext);
+						Mono<State<S, E>> toState = followLinkedPseudoStates(toStateOrig, stateContext);
 						// TODO: try to find matching transition based on direct link.
 						// should make this built-in in pseudostates
-						Transition<S, E> transition = findTransition(toStateOrig, toState);
-						switchToState(toState, null, transition, getRelayStateMachine()).subscribe();
-						pseudoState.exit(stateContext);
+						toState
+							.flatMap(toState2 -> {
+								return Mono.defer(() -> {
+									Transition<S, E> t = findTransition(toStateOrig, toState2);
+									return switchToState(toState2, null, t, getRelayStateMachine());
+								});
+							})
+							.then()
+							.and(pseudoState.exit(stateContext))
+							// TODO: REACTOR should remove fire and forget sub
+							.subscribe();
 					}
 				});
 				// setting instead adding makes sure existing listeners are removed