From d882c392a9c67dfcfe6813f355f68714360f0d16 Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Mon, 31 Aug 2015 10:20:59 +0100 Subject: [PATCH] Fix event handling with threading - With a combination of a task executor using a thread pool, substates with triggerless transitions and actions may fail to properly pass on message headers. - Move queue message outside of loop on a class level which should give next loop a change to see previous queued message which is then used from triggerless transitions. - Relates to #100 --- .../support/DefaultStateMachineExecutor.java | 18 +- .../TransitionEventHeaderTests.java | 310 +++++++++++++----- 2 files changed, 249 insertions(+), 79 deletions(-) diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java index 11e45d0f..1d2067a3 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java @@ -93,6 +93,8 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im private volatile Message forwardedInitialEvent; + private volatile Message queuedMessage = null; + /** * Instantiates a new default state machine executor. * @@ -245,7 +247,9 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im } private void processEventQueue() { - log.debug("Process event queue"); + if (log.isDebugEnabled()) { + log.debug("Process event queue, size=" + eventQueue.size()); + } Message queuedEvent = null; State currentState = stateMachine.getState(); while ((queuedEvent = eventQueue.poll()) != null) { @@ -285,11 +289,12 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im } return; } - log.debug("Process trigger queue"); + if (log.isDebugEnabled()) { + log.debug("Process trigger queue, size=" + triggerQueue.size()); + } TriggerQueueItem queueItem = null; // keep last message here so that we can // pass it to triggerless transitions - Message queuedMessage = null; while ((queueItem = triggerQueue.poll()) != null) { State currentState = stateMachine.getState(); @@ -298,6 +303,9 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im continue; } + // queued message is kept on a class level order to let + // triggerless transition to receive this message if it doesn't + // kick in in this poll loop. queuedMessage = queueItem.message; E event = queuedMessage != null ? queuedMessage.getPayload() : null; @@ -344,7 +352,9 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im } private boolean processDeferList() { - log.debug("Process defer list"); + if (log.isDebugEnabled()) { + log.debug("Process defer list, size=" + deferList.size()); + } boolean triggered = false; ListIterator> iterator = deferList.listIterator(); State currentState = stateMachine.getState(); diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionEventHeaderTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionEventHeaderTests.java index 7e9cf222..727c69ab 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionEventHeaderTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionEventHeaderTests.java @@ -21,28 +21,34 @@ import static org.hamcrest.Matchers.nullValue; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.core.task.TaskExecutor; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.statemachine.AbstractStateMachineTests; import org.springframework.statemachine.ObjectStateMachine; import org.springframework.statemachine.StateContext; +import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.StateMachineSystemConstants; import org.springframework.statemachine.action.Action; import org.springframework.statemachine.config.EnableStateMachine; import org.springframework.statemachine.config.EnumStateMachineConfigurerAdapter; +import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer; import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; import org.springframework.statemachine.guard.Guard; -import org.springframework.statemachine.transition.TransitionTests.TestListener; +import org.springframework.statemachine.listener.StateMachineListenerAdapter; +import org.springframework.statemachine.state.State; /** - * Tests for making sure that events are passed through various - * transition stages. + * Tests for making sure that events are passed through various transition + * stages. * * @author Janne Valkealahti * @@ -61,8 +67,8 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { context.refresh(); assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); - ObjectStateMachine machine = - context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + ObjectStateMachine machine = context.getBean( + StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); EventCheckAction eventCheckAction1 = context.getBean("eventCheckAction1", EventCheckAction.class); EventCheckAction eventCheckAction2 = context.getBean("eventCheckAction2", EventCheckAction.class); @@ -94,12 +100,13 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { context.refresh(); assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); - ObjectStateMachine machine = - context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + ObjectStateMachine machine = context.getBean( + StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); EventCheckAction eventCheckAction1 = context.getBean("eventCheckAction1", EventCheckAction.class); EventCheckAction eventCheckAction3 = context.getBean("eventCheckAction3", EventCheckAction.class); EventCheckAction eventCheckAction4 = context.getBean("eventCheckAction4", EventCheckAction.class); + EventCheckGuard eventCheckGuard = context.getBean("eventCheckGuard", EventCheckGuard.class); TestListener listener = new TestListener(); machine.addStateListener(listener); @@ -116,6 +123,7 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { assertThat(eventCheckAction1.context.getEvent(), nullValue()); assertThat(eventCheckAction3.context, nullValue()); assertThat(eventCheckAction4.context.getEvent(), is(TestEvents.E1)); + assertThat(eventCheckGuard.context.getEvent(), is(TestEvents.E1)); } @SuppressWarnings("unchecked") @@ -125,12 +133,13 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { context.refresh(); assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); - ObjectStateMachine machine = - context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + ObjectStateMachine machine = context.getBean( + StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); EventCheckAction eventCheckAction1 = context.getBean("eventCheckAction1", EventCheckAction.class); EventCheckAction eventCheckAction3 = context.getBean("eventCheckAction3", EventCheckAction.class); EventCheckAction eventCheckAction4 = context.getBean("eventCheckAction4", EventCheckAction.class); + EventCheckGuard eventCheckGuard = context.getBean("eventCheckGuard", EventCheckGuard.class); TestListener listener = new TestListener(); machine.addStateListener(listener); @@ -147,6 +156,76 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { assertThat(eventCheckAction1.context.getEvent(), nullValue()); assertThat(eventCheckAction3.context.getEvent(), is(TestEvents.E1)); assertThat(eventCheckAction4.context, nullValue()); + assertThat(eventCheckGuard.context.getEvent(), is(TestEvents.E1)); + } + + @SuppressWarnings("unchecked") + @Test + public void testEventPassedThroughChoice3() throws Exception { + context.register(Config4.class); + context.refresh(); + + assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); + ObjectStateMachine machine = context.getBean( + StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + + EventCheckAction eventCheckAction1 = context.getBean("eventCheckAction1", EventCheckAction.class); + EventCheckAction eventCheckAction2 = context.getBean("eventCheckAction2", EventCheckAction.class); + EventCheckAction eventCheckAction4 = context.getBean("eventCheckAction4", EventCheckAction.class); + EventCheckGuard eventCheckGuard = context.getBean("eventCheckGuard", EventCheckGuard.class); + + TestListener listener = new TestListener(); + machine.addStateListener(listener); + + machine.start(); + assertThat(machine.getState().getIds(), contains(TestStates.S1)); + + listener.reset(2); + machine.sendEvent(MessageBuilder.withPayload(TestEvents.E1).setHeader("foo", "bar").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(2)); + assertThat(machine.getState().getIds(), contains(TestStates.S4)); + + assertThat(eventCheckAction1.context.getEvent(), nullValue()); + assertThat(eventCheckAction2.context.getEvent(), is(TestEvents.E1)); + assertThat(eventCheckAction4.context.getEvent(), is(TestEvents.E1)); + assertThat(eventCheckGuard.context.getEvent(), is(TestEvents.E1)); + assertThat(eventCheckGuard.context.getMessageHeaders().get("foo", String.class), is("bar")); + } + + @SuppressWarnings("unchecked") + @Test + public void testEventPassedThroughChoice4() throws Exception { + context.register(Config5.class); + context.refresh(); + + assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); + ObjectStateMachine machine = context.getBean( + StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + + EventCheckAction eventCheckAction2 = context.getBean("eventCheckAction2", EventCheckAction.class); + EventCheckAction eventCheckAction20 = context.getBean("eventCheckAction20", EventCheckAction.class); + EventCheckAction eventCheckAction212 = context.getBean("eventCheckAction212", EventCheckAction.class); + EventCheckGuard eventCheckGuard = context.getBean("eventCheckGuard", EventCheckGuard.class); + + TestListener listener = new TestListener(); + machine.addStateListener(listener); + + machine.start(); + assertThat(listener.stateMachineStartedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(machine.getState().getIds(), contains(TestStates.S1)); + + listener.reset(3); + machine.sendEvent(MessageBuilder.withPayload(TestEvents.E1).setHeader("foo", "bar").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(3)); + assertThat(machine.getState().getIds(), contains(TestStates.S2, TestStates.S212)); + + assertThat(eventCheckAction2.context.getEvent(), is(TestEvents.E1)); + assertThat(eventCheckAction20.context.getEvent(), is(TestEvents.E1)); + assertThat(eventCheckAction212.context.getEvent(), is(TestEvents.E1)); + assertThat(eventCheckGuard.context.getEvent(), is(TestEvents.E1)); + assertThat(eventCheckGuard.context.getMessageHeaders().get("foo", String.class), is("bar")); } @Configuration @@ -155,30 +234,16 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { @Override public void configure(StateMachineStateConfigurer states) throws Exception { - states - .withStates() - .initial(TestStates.S1) - .state(TestStates.S1, eventCheckAction1(), null) - .state(TestStates.S2, eventCheckAction2(), null) - .state(TestStates.S3, eventCheckAction3(), null) + states.withStates().initial(TestStates.S1).state(TestStates.S1, eventCheckAction1(), null) + .state(TestStates.S2, eventCheckAction2(), null).state(TestStates.S3, eventCheckAction3(), null) .state(TestStates.S4, eventCheckAction4(), null); } @Override public void configure(StateMachineTransitionConfigurer transitions) throws Exception { - transitions - .withExternal() - .source(TestStates.S1) - .target(TestStates.S2) - .event(TestEvents.E1) - .and() - .withExternal() - .source(TestStates.S2) - .target(TestStates.S3) - .and() - .withExternal() - .source(TestStates.S3) - .target(TestStates.S4); + transitions.withExternal().source(TestStates.S1).target(TestStates.S2).event(TestEvents.E1).and() + .withExternal().source(TestStates.S2).target(TestStates.S3).and().withExternal() + .source(TestStates.S3).target(TestStates.S4); } @Bean @@ -208,27 +273,15 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { @Override public void configure(StateMachineStateConfigurer states) throws Exception { - states - .withStates() - .initial(TestStates.S1) - .state(TestStates.S1, eventCheckAction1(), null) - .choice(TestStates.S2) - .state(TestStates.S3, eventCheckAction3(), null) + states.withStates().initial(TestStates.S1).state(TestStates.S1, eventCheckAction1(), null) + .choice(TestStates.S2).state(TestStates.S3, eventCheckAction3(), null) .state(TestStates.S4, eventCheckAction4(), null); } @Override public void configure(StateMachineTransitionConfigurer transitions) throws Exception { - transitions - .withExternal() - .source(TestStates.S1) - .target(TestStates.S2) - .event(TestEvents.E1) - .and() - .withChoice() - .source(TestStates.S2) - .first(TestStates.S3, guard()) - .last(TestStates.S4); + transitions.withExternal().source(TestStates.S1).target(TestStates.S2).event(TestEvents.E1).and() + .withChoice().source(TestStates.S2).first(TestStates.S3, eventCheckGuard()).last(TestStates.S4); } @Bean @@ -247,14 +300,8 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { } @Bean - public Guard guard() { - return new Guard() { - - @Override - public boolean evaluate(StateContext context) { - return false; - } - }; + public EventCheckGuard eventCheckGuard() { + return new EventCheckGuard(false); } } @@ -264,27 +311,15 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { @Override public void configure(StateMachineStateConfigurer states) throws Exception { - states - .withStates() - .initial(TestStates.S1) - .state(TestStates.S1, eventCheckAction1(), null) - .choice(TestStates.S2) - .state(TestStates.S3, eventCheckAction3(), null) + states.withStates().initial(TestStates.S1).state(TestStates.S1, eventCheckAction1(), null) + .choice(TestStates.S2).state(TestStates.S3, eventCheckAction3(), null) .state(TestStates.S4, eventCheckAction4(), null); } @Override public void configure(StateMachineTransitionConfigurer transitions) throws Exception { - transitions - .withExternal() - .source(TestStates.S1) - .target(TestStates.S2) - .event(TestEvents.E1) - .and() - .withChoice() - .source(TestStates.S2) - .first(TestStates.S3, guard()) - .last(TestStates.S4); + transitions.withExternal().source(TestStates.S1).target(TestStates.S2).event(TestEvents.E1).and() + .withChoice().source(TestStates.S2).first(TestStates.S3, eventCheckGuard()).last(TestStates.S4); } @Bean @@ -303,14 +338,116 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { } @Bean - public Guard guard() { - return new Guard() { + public EventCheckGuard eventCheckGuard() { + return new EventCheckGuard(true); + } + } - @Override - public boolean evaluate(StateContext context) { - return true; - } - }; + @Configuration + @EnableStateMachine + public static class Config4 extends EnumStateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states.withStates().initial(TestStates.S1).state(TestStates.S1, eventCheckAction1(), null) + .state(TestStates.S2, eventCheckAction2(), null).choice(TestStates.S3) + .state(TestStates.S4, eventCheckAction4(), null); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions.withExternal().source(TestStates.S1).target(TestStates.S2).event(TestEvents.E1).and() + .withExternal().source(TestStates.S2).target(TestStates.S3).and().withChoice() + .source(TestStates.S3).first(TestStates.S4, eventCheckGuard()).last(TestStates.S4); + } + + @Bean + public EventCheckAction eventCheckAction1() { + return new EventCheckAction(); + } + + @Bean + public EventCheckAction eventCheckAction2() { + return new EventCheckAction(); + } + + @Bean + public EventCheckAction eventCheckAction4() { + return new EventCheckAction(); + } + + @Bean + public EventCheckGuard eventCheckGuard() { + return new EventCheckGuard(true); + } + } + + @Configuration + @EnableStateMachine + public static class Config5 extends EnumStateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineConfigurationConfigurer config) throws Exception { + config.withConfiguration().taskExecutor(customTaskExecutor()); + } + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states.withStates().initial(TestStates.S1).state(TestStates.S2, eventCheckAction2(), null).and() + .withStates().parent(TestStates.S2).initial(TestStates.S20) + .state(TestStates.S20, eventCheckAction20(), null).choice(TestStates.S211) + .state(TestStates.S212, eventCheckAction212(), null); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions.withExternal().source(TestStates.S1).target(TestStates.S2).event(TestEvents.E1).and() + .withExternal().source(TestStates.S20).target(TestStates.S211).and().withChoice() + .source(TestStates.S211).first(TestStates.S212, eventCheckGuard()).last(TestStates.S212); + } + + @Bean + public EventCheckAction eventCheckAction2() { + return new EventCheckAction(); + } + + @Bean + public EventCheckAction eventCheckAction20() { + return new EventCheckAction(); + } + + @Bean + public EventCheckAction eventCheckAction212() { + return new EventCheckAction(); + } + + @Bean + public EventCheckGuard eventCheckGuard() { + return new EventCheckGuard(true); + } + + @Bean(name = StateMachineSystemConstants.TASK_EXECUTOR_BEAN_NAME) + public TaskExecutor customTaskExecutor() { + ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); + taskExecutor.setCorePoolSize(1); + return taskExecutor; + } + + } + + private static class EventCheckGuard implements Guard { + + boolean ret = false; + StateContext context; + + public EventCheckGuard(boolean ret) { + this.ret = ret; + } + + @Override + public boolean evaluate(StateContext context) { + this.context = context; + return ret; } } @@ -322,6 +459,29 @@ public class TransitionEventHeaderTests extends AbstractStateMachineTests { public void execute(StateContext context) { this.context = context; } + } + + static class TestListener extends StateMachineListenerAdapter { + + volatile CountDownLatch stateMachineStartedLatch = new CountDownLatch(1); + volatile CountDownLatch stateChangedLatch = new CountDownLatch(1); + volatile int stateChangedCount = 0; + + @Override + public void stateMachineStarted(StateMachine stateMachine) { + stateMachineStartedLatch.countDown(); + } + + @Override + public void stateChanged(State from, State to) { + stateChangedCount++; + stateChangedLatch.countDown(); + } + + public void reset(int c1) { + stateChangedLatch = new CountDownLatch(c1); + stateChangedCount = 0; + } }