- Deprecate handleEventWithState in favour of handleEventWithStateReactively.

- Change handleEventWithState to use blocking methods as function
  is essentially blocking. Will get removed in future.
- Fix tests
- Relates #750

spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java

@@ -30,6 +30,8 @@ import org.springframework.statemachine.support.StateMachineInterceptorAdapter;
 import org.springframework.statemachine.transition.Transition;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Mono;
+
 /**
  * {@code PersistStateMachineHandler} is a recipe which can be used to
  * handle a state change of an arbitrary entity in a persistent storage.
@@ -70,17 +72,40 @@ public class PersistStateMachineHandler extends LifecycleObjectSupport {
 	 * @param event the event
 	 * @param state the state
 	 * @return true if event was accepted
+	 * @see #handleEventWithStateReactively(Message, String)
 	 */
+	@Deprecated
 	public boolean handleEventWithState(Message<String> event, String state) {
-		stateMachine.stop();
+		stateMachine.stopReactively().block();
 		List<StateMachineAccess<String, String>> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions();
 		for (StateMachineAccess<String, String> a : withAllRegions) {
 			a.resetStateMachine(new DefaultStateMachineContext<String, String>(state, null, null, null));
 		}
-		stateMachine.start();
+		stateMachine.startReactively().block();
 		return stateMachine.sendEvent(event);
 	}
 
+	/**
+	 * Handle event with entity reactively.
+	 *
+	 * @param event the event
+	 * @param state the state
+	 * @return mono for completion
+	 */
+	public Mono<Void> handleEventWithStateReactively(Message<String> event, String state) {
+		// TODO: REACTOR add docs and revisit this function concept
+		return Mono.from(stateMachine.stopReactively())
+			.then(Mono.fromRunnable(() -> {
+				List<StateMachineAccess<String, String>> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions();
+				for (StateMachineAccess<String, String> a : withAllRegions) {
+					a.resetStateMachine(new DefaultStateMachineContext<String, String>(state, null, null, null));
+				}
+			}))
+			.then(stateMachine.startReactively())
+			.thenMany(stateMachine.sendEvent(Mono.just(event)))
+			.then();
+	}
+
 	/**
 	 * Adds the persist state change listener.
 	 *

spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/tasks/TasksHandler.java

@@ -26,6 +26,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.statemachine.StateContext;
 import org.springframework.statemachine.StateMachine;
@@ -52,6 +53,8 @@ import org.springframework.statemachine.support.tree.Tree.Node;
 import org.springframework.statemachine.support.tree.TreeTraverser;
 import org.springframework.statemachine.transition.Transition;
 
+import reactor.core.publisher.Mono;
+
 /**
  * {@code TasksHandler} is a recipe for executing arbitrary {@link Runnable} tasks
  * using a state machine logic.
@@ -124,21 +127,30 @@ public class TasksHandler {
 	 * Request to execute current tasks logic.
 	 */
 	public void runTasks() {
-		stateMachine.sendEvent(EVENT_RUN);
+		stateMachine
+			.sendEvent(Mono.just(MessageBuilder
+				.withPayload(EVENT_RUN).build()))
+			.subscribe();
 	}
 
 	/**
 	 * Request to continue from an error.
 	 */
 	public void continueFromError() {
-		stateMachine.sendEvent(EVENT_CONTINUE);
+		stateMachine
+			.sendEvent(Mono.just(MessageBuilder
+				.withPayload(EVENT_CONTINUE).build()))
+			.subscribe();
 	}
 
 	/**
 	 * Request to fix current problems.
 	 */
 	public void fixCurrentProblems() {
-		stateMachine.sendEvent(EVENT_FIX);
+		stateMachine
+			.sendEvent(Mono.just(MessageBuilder
+				.withPayload(EVENT_FIX).build()))
+			.subscribe();
 	}
 
 	/**
@@ -160,7 +172,7 @@ public class TasksHandler {
 			throw new StateMachineException("Error reading state from persistent store", e);
 		}
 
-		stateMachine.stop();
+		stateMachine.stopReactively().block();
 		stateMachine.getStateMachineAccessor()
 			.doWithAllRegions(new StateMachineFunction<StateMachineAccess<String, String>>() {
 
@@ -169,7 +181,7 @@ public class TasksHandler {
 				function.resetStateMachine(context);
 			}
 		});
-		stateMachine.start();
+		stateMachine.startReactively().block();
 	}
 
 	/**
@@ -535,9 +547,15 @@ public class TasksHandler {
 					}
 				}
 				if (hasErrors) {
-					context.getStateMachine().sendEvent(EVENT_FALLBACK);
+					context.getStateMachine()
+						.sendEvent(Mono.just(MessageBuilder
+							.withPayload(EVENT_FALLBACK).build()))
+						.subscribe();
 				} else {
-					context.getStateMachine().sendEvent(EVENT_CONTINUE);
+					context.getStateMachine()
+						.sendEvent(Mono.just(MessageBuilder
+							.withPayload(EVENT_CONTINUE).build()))
+						.subscribe();
 				}
 			}
 		};

spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/PersistStateMachineHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2019 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.
@@ -38,6 +38,7 @@ import org.springframework.statemachine.transition.Transition;
 public class PersistStateMachineHandlerTests {
 
 	@Test
+	@SuppressWarnings("deprecation")
 	public void testAcceptedStateChangeViaPersist() throws Exception {
 		StateMachine<String,String> stateMachine = buildTestStateMachine();
 
@@ -56,6 +57,7 @@ public class PersistStateMachineHandlerTests {
 	}
 
 	@Test
+	@SuppressWarnings("deprecation")
 	public void testNotAcceptedStateChangeViaPersist() throws Exception {
 		StateMachine<String,String> stateMachine = buildTestStateMachine();
 
@@ -74,6 +76,7 @@ public class PersistStateMachineHandlerTests {
 	}
 
 	@Test
+	@SuppressWarnings("deprecation")
 	public void testChoice() throws Exception {
 		StateMachine<String,String> stateMachine = buildTestStateMachine2();
 
@@ -94,6 +97,24 @@ public class PersistStateMachineHandlerTests {
 		assertThat(stateMachine.getState().getIds(), containsInAnyOrder("S2"));
 	}
 
+	@Test
+	public void testAcceptedStateChangeViaPersistReactively() throws Exception {
+		StateMachine<String,String> stateMachine = buildTestStateMachine();
+
+		PersistStateMachineHandler handler = new PersistStateMachineHandler(stateMachine);
+		handler.afterPropertiesSet();
+		handler.start();
+
+		TestPersistStateChangeListener listener = new TestPersistStateChangeListener();
+		handler.addPersistStateChangeListener(listener);
+
+		Message<String> event = MessageBuilder.withPayload("E2").build();
+		handler.handleEventWithStateReactively(event, "S1").subscribe();
+
+		assertThat(listener.latch.await(1, TimeUnit.SECONDS), is(true));
+		assertThat(stateMachine.getState().getIds(), containsInAnyOrder("S2"));
+	}
+
 	private static class TestPersistStateChangeListener implements PersistStateChangeListener {
 
 		CountDownLatch latch = new CountDownLatch(1);