Clear event defer list

- In ReactiveStateMachineExecutor deferList is clear
  with deferred event if it triggers.
- Fixes #747

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

@@ -54,6 +54,14 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 
+/**
+ * Default reactive implementation of a {@link StateMachineExecutor}.
+ * 
+ * @author Janne Valkealahti
+ *
+ * @param <S> the type of state
+ * @param <E> the type of event
+ */
 public class ReactiveStateMachineExecutor<S, E> extends LifecycleObjectSupport implements StateMachineExecutor<S, E> {
 
 	private static final Log log = LogFactory.getLog(ReactiveStateMachineExecutor.class);
@@ -66,15 +74,12 @@ public class ReactiveStateMachineExecutor<S, E> extends LifecycleObjectSupport i
 	private final Message<E> initialEvent;
 	private final TransitionComparator<S, E> transitionComparator;
 	private final TransitionConflictPolicy transitionConflictPolicy;
-	// TODO deferList is never cleared
 	private final Queue<Message<E>> deferList = new ConcurrentLinkedQueue<Message<E>>();
 	private final AtomicBoolean initialHandled = new AtomicBoolean(false);
 	private final StateMachineInterceptorList<S, E> interceptors = new StateMachineInterceptorList<S, E>();
-
 	private volatile Message<E> forwardedInitialEvent;
 	private volatile Message<E> queuedMessage = null;
 	private StateMachineExecutorTransit<S, E> stateMachineExecutorTransit;
-
 	private EmitterProcessor<TriggerQueueItem> triggerProcessor = EmitterProcessor.create(false);
 	private FluxSink<TriggerQueueItem> triggerSink;
 	private Flux<Void> triggerFlux;
@@ -215,6 +220,7 @@ public class ReactiveStateMachineExecutor<S, E> extends LifecycleObjectSupport i
 
 				if (StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) {
 					if (trigger != null && trigger.evaluate(new DefaultTriggerContext<S, E>(queuedEvent.getPayload()))) {
+						deferList.remove(queuedEvent);
 						return Mono.just(new TriggerQueueItem(trigger, queuedEvent));
 					}
 				}

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

@@ -365,6 +365,32 @@ public class EventDeferTests extends AbstractStateMachineTests {
 		assertThat(machine.getState().getIds(), contains("SUB2"));
 	}
 
+	@Test
+	public void testDeferEventCleared() throws Exception {
+		context.register(Config5.class);
+		context.refresh();
+		@SuppressWarnings("unchecked")
+		StateMachine<String, String> machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class);
+
+		machine.start();
+		assertThat(machine.getState().getIds(), containsInAnyOrder("S1"));
+
+		machine.sendEvent("E2");
+		assertThat(machine.getState().getIds(), containsInAnyOrder("S1"));
+		Object executor = TestUtils.readField("stateMachineExecutor", machine);
+		Collection<?> readField = TestUtils.readField("deferList", executor);
+		assertThat(readField.size(), is(1));
+
+		machine.sendEvent("E1");
+		assertThat(machine.getState().getIds(), containsInAnyOrder("S1"));
+		readField = TestUtils.readField("deferList", executor);
+		assertThat(readField.size(), is(0));
+
+		// deferred event handled so should not get back to S1
+		machine.sendEvent("E1");
+		assertThat(machine.getState().getIds(), containsInAnyOrder("S2"));
+	}
+
 	@Configuration
 	@EnableStateMachine
 	static class Config1 extends StateMachineConfigurerAdapter<String, String> {