Make emit non-blocking

- Now doing triggerSink.emitNext with retry instead
  of parking which caused blocking errors from blockhound.
- Retry is now hardcoded and settings for it could be
  exposed later.

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.statemachine.support;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -28,7 +29,6 @@ import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.LockSupport;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -57,9 +57,11 @@ import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
+import reactor.core.publisher.Sinks.EmitFailureHandler;
 import reactor.core.publisher.Sinks.Many;
 import reactor.util.concurrent.Queues;
 import reactor.util.context.Context;
+import reactor.util.retry.Retry;
 
 /**
  * Default reactive implementation of a {@link StateMachineExecutor}.
@@ -152,17 +154,6 @@ public class ReactiveStateMachineExecutor<S, E> extends LifecycleObjectSupport i
 		return stopTriggers().and(mono);
 	}
 
-	@Override
-	public void queueTrigger(Trigger<S, E> trigger, Message<E> message) {
-		if (log.isDebugEnabled()) {
-			log.debug("Queue trigger " + trigger);
-		}
-		TriggerQueueItem tqi = new TriggerQueueItem(trigger, message, null, null);
-		while (triggerSink.tryEmitNext(tqi).isFailure()) {
-			LockSupport.parkNanos(10);
-		}
-	}
-
 	@Override
 	public void queueDeferredEvent(Message<E> message) {
 		if (log.isDebugEnabled()) {
@@ -211,11 +202,10 @@ public class ReactiveStateMachineExecutor<S, E> extends LifecycleObjectSupport i
 
 		return messages
 			.flatMap(m -> handleEvent(m, callback, triggerCallback))
-			.doOnNext(i -> {
-				while (triggerSink.tryEmitNext(i).isFailure()) {
-					LockSupport.parkNanos(10);
-				}
-			})
+			.flatMap(tqi -> Mono.fromRunnable(() -> {
+					triggerSink.emitNext(tqi, EmitFailureHandler.FAIL_FAST);
+				})
+				.retryWhen(Retry.fixedDelay(10, Duration.ofMillis(10))))
 			.then()
 			.and(triggerCallbackSink);
 	}
@@ -464,7 +454,13 @@ public class ReactiveStateMachineExecutor<S, E> extends LifecycleObjectSupport i
 						if (log.isDebugEnabled()) {
 							log.debug("TimedTrigger triggered " + trigger);
 						}
-						queueTrigger(trigger, null);
+						Mono.just(new TriggerQueueItem(trigger, null, null, null))
+							.flatMap(tqi -> Mono.fromCallable(() -> {
+									triggerSink.emitNext(tqi, EmitFailureHandler.FAIL_FAST);
+									return null;
+								})
+								.retryWhen(Retry.fixedDelay(10, Duration.ofNanos(10))))
+							.subscribe();
 					}
 				});
 			}
@@ -514,5 +510,10 @@ public class ReactiveStateMachineExecutor<S, E> extends LifecycleObjectSupport i
 			this.callback = callback;
 			this.triggerCallback = triggerCallback;
 		}
+
+		@Override
+		public String toString() {
+			return "TriggerQueueItem [message=" + message + ", trigger=" + trigger + "]";
+		}
 	}
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.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.
@@ -23,7 +23,6 @@ import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.access.StateMachineAccess;
 import org.springframework.statemachine.state.State;
 import org.springframework.statemachine.transition.Transition;
-import org.springframework.statemachine.trigger.Trigger;
 
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
@@ -47,14 +46,6 @@ public interface StateMachineExecutor<S, E> extends StateMachineReactiveLifecycl
 	 */
 	Mono<Void> queueEvent(Mono<Message<E>> message, StateMachineExecutorCallback callback);
 
-	/**
-	 * Queue trigger.
-	 *
-	 * @param trigger the trigger
-	 * @param message the message
-	 */
-	void queueTrigger(Trigger<S, E> trigger, Message<E> message);
-
 	/**
 	 * Queue deferred event.
 	 *

spring-statemachine-core/src/test/java/org/springframework/statemachine/EventDeferTests.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.
@@ -30,7 +30,10 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.statemachine.config.EnableStateMachine;
@@ -42,6 +45,8 @@ import org.springframework.statemachine.state.State;
 
 public class EventDeferTests extends AbstractStateMachineTests {
 
+	private static final Log log = LogFactory.getLog(EventDeferTests.class);
+
 	@Override
 	protected AnnotationConfigApplicationContext buildContext() {
 		return new AnnotationConfigApplicationContext();
@@ -65,6 +70,7 @@ public class EventDeferTests extends AbstractStateMachineTests {
 	}
 
 	@Test
+	@Timeout(value = 20, unit = TimeUnit.SECONDS)
 	public void testDeferSmokeExecutorConcurrentModification() throws Exception {
 		context.register(Config5.class);
 		context.refresh();
@@ -73,20 +79,15 @@ public class EventDeferTests extends AbstractStateMachineTests {
 		doStartAndAssert(machine);
 		assertThat(machine.getState().getIds(), contains("S1"));
 
-		doSendEventAndConsumeAll(machine, "E2");
-
-		Object executor = TestUtils.readField("stateMachineExecutor", machine);
-		Collection<?> readField = TestUtils.readField("deferList", executor);
-		assertThat(readField.size(), is(1));
-
-		AtomicReference<Exception> error = new AtomicReference<>();
+		AtomicReference<Throwable> error = new AtomicReference<>();
 		AtomicInteger i1 = new AtomicInteger();
 		Thread t1 = new Thread(() -> {
 			while(i1.incrementAndGet() < 200) {
 				try {
 					doSendEventAndConsumeAll(machine, "E1");
 					doSendEventAndConsumeAll(machine, "E2");
-				} catch (Exception e) {
+				} catch (Throwable e) {
+					log.error("T1: error " + e);
 					error.set(e);
 					break;
 				}
@@ -98,7 +99,8 @@ public class EventDeferTests extends AbstractStateMachineTests {
 				try {
 					doSendEventAndConsumeAll(machine, "E1");
 					doSendEventAndConsumeAll(machine, "E2");
-				} catch (Exception e) {
+				} catch (Throwable e) {
+					log.error("T2: error " + e);
 					error.set(e);
 					break;
 				}

spring-statemachine-core/src/test/java/org/springframework/statemachine/ReactiveTests.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.
@@ -21,8 +21,10 @@ import static org.hamcrest.Matchers.containsInAnyOrder;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.messaging.Message;
@@ -60,6 +62,7 @@ public class ReactiveTests extends AbstractStateMachineTests {
 
 	@SuppressWarnings("unchecked")
 	@Test
+	@Timeout(value = 10, unit = TimeUnit.SECONDS)
 	public void testMonosAllAccepted() {
 		context.register(Config1.class);
 		context.refresh();

spring-statemachine-core/src/test/java/org/springframework/statemachine/TestUtils.java

@@ -21,6 +21,7 @@ import static org.springframework.statemachine.StateMachineSystemConstants.DEFAU
 
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
+import java.time.Duration;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.messaging.Message;
@@ -99,7 +100,8 @@ public class TestUtils {
 	public static <S, E> void doSendEventAndConsumeAll(StateMachine<S, E> stateMachine, E event) {
 		StepVerifier.create(stateMachine.sendEvent(eventAsMono(event)))
 			.thenConsumeWhile(eventResult -> true)
-			.verifyComplete();
+			.expectComplete()
+			.verify(Duration.ofSeconds(5));
 	}
 
 	public static <S, E> void doSendEventAndConsumeAll(StateMachine<S, E> stateMachine, Message<E> event) {