diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java index c04a9c4f..439de653 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java @@ -64,6 +64,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.locks.Lock; /** * Base implementation of a {@link StateMachine} loosely modelled from UML state @@ -178,12 +179,11 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo public State getState() { // if we're complete assume we're stopped // and state was stashed into lastState - synchronized (lock) { - if (lastState != null && isComplete()) { - return lastState; - } else { - return currentState; - } + State s = lastState; + if (s != null && isComplete()) { + return s; + } else { + return currentState; } } @@ -286,6 +286,19 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo @Override public void transit(Transition t, StateContext ctx, Message message) { + if (currentState != null && currentState.isSubmachineState()) { + // this is a naive attempt to check from submachine's executor if it is + // currently executing. allows submachine to complete its execution logic + // before we, in parent go forward. as executor locks, we simple try to lock it + // and release it immediately. + StateMachine submachine = ((AbstractState)currentState).getSubmachine(); + Lock lock = ((AbstractStateMachine)submachine).getStateMachineExecutor().getLock(); + try { + lock.lock(); + } finally { + lock.unlock(); + } + } long now = System.currentTimeMillis(); // TODO: fix above stateContext as it's not used notifyTransitionStart(buildStateContext(Stage.TRANSITION_START, message, t, getRelayStateMachine())); @@ -334,6 +347,10 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo } } + protected StateMachineExecutor getStateMachineExecutor() { + return stateMachineExecutor; + } + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { // last change to set factory because this maybe be called per 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 65dd4d40..60518962 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 @@ -30,6 +30,8 @@ import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -97,6 +99,8 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im private volatile Message queuedMessage = null; + private final ReentrantLock lock = new ReentrantLock(); + /** * Instantiates a new default state machine executor. * @@ -183,6 +187,11 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im interceptors.add(interceptor); } + @Override + public Lock getLock() { + return lock; + } + private final Set> joinSyncTransitions = new HashSet<>(); private final Set> joinSyncStates = new HashSet<>(); @@ -274,29 +283,37 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im // TODO: it'd be nice not to create runnable if // current ref is null, we use atomic reference - // to play safe with concurrency + // to play safe with concurrency. Runnable task = new Runnable() { @Override public void run() { - boolean eventProcessed = false; - while (processEventQueue()) { - eventProcessed = true; - processTriggerQueue(); - while (processDeferList()) { + // lock operation, see AbstractStateMachine + // how this is used. + lock.lock(); + try { + boolean eventProcessed = false; + while (processEventQueue()) { + eventProcessed = true; processTriggerQueue(); + while (processDeferList()) { + processTriggerQueue(); + } } - } - if (!eventProcessed) { - processTriggerQueue(); - while (processDeferList()) { + if (!eventProcessed) { processTriggerQueue(); + while (processDeferList()) { + processTriggerQueue(); + } } + + if (requestTask.getAndSet(false)) { + scheduleEventQueueProcessing(); + } + taskRef.set(null); + } finally { + lock.unlock(); } - if (requestTask.getAndSet(false)) { - scheduleEventQueueProcessing(); - } - taskRef.set(null); // do second attempt which should reduse risk // of threading causing failed run to completion if (requestTask.getAndSet(false)) { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java index 7f3951e4..e39560fe 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 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. @@ -15,6 +15,8 @@ */ package org.springframework.statemachine.support; +import java.util.concurrent.locks.Lock; + import org.springframework.messaging.Message; import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; @@ -102,6 +104,13 @@ public interface StateMachineExecutor { */ void addStateMachineInterceptor(StateMachineInterceptor interceptor); + /** + * Gets the execution lock. + * + * @return the execution lock + */ + Lock getLock(); + /** * Callback interface when executor wants to handle transit. */ diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/EventHeaderTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/EventHeaderTests.java index 67fa1b83..49592307 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/EventHeaderTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/EventHeaderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 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. @@ -26,7 +26,9 @@ 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.action.Action; import org.springframework.statemachine.config.EnableStateMachine; import org.springframework.statemachine.config.StateMachineConfigurerAdapter; @@ -179,6 +181,199 @@ public class EventHeaderTests extends AbstractStateMachineTests { assertThat(headerTestAction112.testHeader, nullValue()); } + @SuppressWarnings("unchecked") + @Test + public void testHeaderPassedToInitialInSubs1Threading() throws InterruptedException { + context.register(Config2.class); + context.refresh(); + StateMachine machine = context.getBean(StateMachine.class); + HeaderTestAction headerTestAction1I = context.getBean("headerTestAction1I", HeaderTestAction.class); + HeaderTestAction headerTestAction1 = context.getBean("headerTestAction1", HeaderTestAction.class); + HeaderTestAction headerTestAction11 = context.getBean("headerTestAction11", HeaderTestAction.class); + HeaderTestAction headerTestAction111 = context.getBean("headerTestAction111", HeaderTestAction.class); + HeaderTestAction headerTestAction112 = context.getBean("headerTestAction112", HeaderTestAction.class); + TestListener listener = new TestListener(); + listener.reset(1); + machine.addStateListener(listener); + machine.start(); + + assertThat(listener.stateMachineStartedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(1)); + + listener.reset(3); + machine.sendEvent(MessageBuilder.withPayload("E1").setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(3)); + + assertThat(headerTestAction1I.testHeader, is("testValue")); + assertThat(headerTestAction1.testHeader, is("testValue")); + assertThat(headerTestAction11.testHeader, is("testValue")); + assertThat(headerTestAction111.testHeader, is("testValue")); + assertThat(headerTestAction112.testHeader, nullValue()); + + headerTestAction1.testHeader = null; + headerTestAction11.testHeader = null; + headerTestAction111.testHeader = null; + + listener.reset(1); + machine.sendEvent(MessageBuilder.withPayload("E2").setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(1)); + + assertThat(headerTestAction1.testHeader, nullValue()); + assertThat(headerTestAction11.testHeader, nullValue()); + assertThat(headerTestAction111.testHeader, nullValue()); + assertThat(headerTestAction112.testHeader, is("testValue")); + } + + @SuppressWarnings("unchecked") + @Test + public void testHeaderPassedToInitialInSubs2Threading() throws InterruptedException { + context.register(Config2.class); + context.refresh(); + StateMachine machine = context.getBean(StateMachine.class); + HeaderTestAction headerTestAction1 = context.getBean("headerTestAction1", HeaderTestAction.class); + HeaderTestAction headerTestAction11 = context.getBean("headerTestAction11", HeaderTestAction.class); + HeaderTestAction headerTestAction111 = context.getBean("headerTestAction111", HeaderTestAction.class); + HeaderTestAction headerTestAction112 = context.getBean("headerTestAction112", HeaderTestAction.class); + TestListener listener = new TestListener(); + listener.reset(1); + machine.addStateListener(listener); + machine.start(); + + assertThat(listener.stateMachineStartedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(1)); + + listener.reset(3); + machine.sendEvent(MessageBuilder.withPayload("E1").setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(3)); + + assertThat(headerTestAction1.testHeader, is("testValue")); + assertThat(headerTestAction11.testHeader, is("testValue")); + assertThat(headerTestAction111.testHeader, is("testValue")); + assertThat(headerTestAction112.testHeader, nullValue()); + + headerTestAction1.testHeader = null; + headerTestAction11.testHeader = null; + headerTestAction111.testHeader = null; + + listener.reset(1); + machine.sendEvent(MessageBuilder.withPayload("E2").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(1)); + + assertThat(headerTestAction1.testHeader, nullValue()); + assertThat(headerTestAction11.testHeader, nullValue()); + assertThat(headerTestAction111.testHeader, nullValue()); + assertThat(headerTestAction112.testHeader, nullValue()); + } + + @SuppressWarnings("unchecked") + @Test + public void testHeaderPassedToInitialInSubs3Threading() throws InterruptedException { + context.register(Config2.class); + context.refresh(); + StateMachine machine = context.getBean(StateMachine.class); + HeaderTestAction headerTestAction1I = context.getBean("headerTestAction1I", HeaderTestAction.class); + HeaderTestAction headerTestAction1 = context.getBean("headerTestAction1", HeaderTestAction.class); + HeaderTestAction headerTestAction11 = context.getBean("headerTestAction11", HeaderTestAction.class); + HeaderTestAction headerTestAction111 = context.getBean("headerTestAction111", HeaderTestAction.class); + HeaderTestAction headerTestAction112 = context.getBean("headerTestAction112", HeaderTestAction.class); + TestListener listener = new TestListener(); + listener.reset(1); + machine.addStateListener(listener); + machine.start(); + + assertThat(listener.stateMachineStartedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(1)); + + listener.reset(3); + machine.sendEvent(MessageBuilder.withPayload("E1").setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(3)); + + listener.reset(1); + machine.sendEvent(MessageBuilder.withPayload("E2").setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(1)); + + headerTestAction1I.testHeader = null; + headerTestAction1.testHeader = null; + headerTestAction11.testHeader = null; + headerTestAction111.testHeader = null; + headerTestAction112.testHeader = null; + listener.reset(1); + machine.sendEvent(MessageBuilder.withPayload("E3").setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(1)); + + assertThat(headerTestAction1I.testHeader, nullValue()); + assertThat(headerTestAction1.testHeader, nullValue()); + assertThat(headerTestAction11.testHeader, nullValue()); + assertThat(headerTestAction111.testHeader, is("testValue")); + assertThat(headerTestAction112.testHeader, nullValue()); + } + + @SuppressWarnings("unchecked") + @Test + public void testHeaderPassedWithAnonymousTransition() throws InterruptedException { + context.register(Config3.class); + context.refresh(); + StateMachine machine = context.getBean(StateMachine.class); + HeaderTestAction headerTestAction1 = context.getBean("headerTestAction1", HeaderTestAction.class); + HeaderTestAction headerTestAction2 = context.getBean("headerTestAction2", HeaderTestAction.class); + HeaderTestAction headerTestAction3 = context.getBean("headerTestAction3", HeaderTestAction.class); + TestListener listener = new TestListener(); + listener.reset(1); + machine.addStateListener(listener); + machine.start(); + + assertThat(listener.stateMachineStartedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(1)); + + listener.reset(3); + machine.sendEvent(MessageBuilder.withPayload("E1").setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(3)); + + assertThat(headerTestAction1.testHeader, is("testValue")); + assertThat(headerTestAction2.testHeader, is("testValue")); + assertThat(headerTestAction3.testHeader, is("testValue")); + } + + @SuppressWarnings("unchecked") + @Test + public void testHeaderPassedWithAnonymousTransitionThreading() throws InterruptedException { + context.register(Config4.class); + context.refresh(); + StateMachine machine = context.getBean(StateMachine.class); + HeaderTestAction headerTestAction1 = context.getBean("headerTestAction1", HeaderTestAction.class); + HeaderTestAction headerTestAction2 = context.getBean("headerTestAction2", HeaderTestAction.class); + HeaderTestAction headerTestAction3 = context.getBean("headerTestAction3", HeaderTestAction.class); + TestListener listener = new TestListener(); + listener.reset(1); + machine.addStateListener(listener); + machine.start(); + + assertThat(listener.stateMachineStartedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(1)); + + listener.reset(3); + machine.sendEvent(MessageBuilder.withPayload("E1").setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(3)); + + assertThat(headerTestAction1.testHeader, is("testValue")); + assertThat(headerTestAction2.testHeader, is("testValue")); + assertThat(headerTestAction3.testHeader, is("testValue")); + } + @Configuration @EnableStateMachine static class Config1 extends StateMachineConfigurerAdapter { @@ -248,6 +443,183 @@ public class EventHeaderTests extends AbstractStateMachineTests { } } + @Configuration + @EnableStateMachine + static class Config2 extends StateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial("SI") + .state("S1", headerTestAction1(), null) + .and() + .withStates() + .parent("S1") + .initial("S11", headerTestAction1I()) + .state("S11", headerTestAction11(), null) + .state("S12") + .and() + .withStates() + .parent("S11") + .initial("S111") + .state("S111", headerTestAction111(), null) + .state("S122", headerTestAction112(), null); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions + .withExternal() + .source("SI") + .target("S1") + .event("E1") + .and() + .withExternal() + .source("S111") + .target("S122") + .event("E2") + .and() + .withExternal() + .source("S122") + .target("S111") + .event("E3"); + } + + @Bean + public HeaderTestAction headerTestAction1I() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction headerTestAction1() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction headerTestAction11() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction headerTestAction111() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction headerTestAction112() { + return new HeaderTestAction(); + } + + @Bean(name = StateMachineSystemConstants.TASK_EXECUTOR_BEAN_NAME) + public TaskExecutor taskExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(1); + return executor; + } + } + + @Configuration + @EnableStateMachine + static class Config3 extends StateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial("SI") + .state("S1", headerTestAction1(), null) + .state("S2", headerTestAction2(), null) + .state("S3", headerTestAction3(), null); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions + .withExternal() + .source("SI") + .target("S1") + .event("E1") + .and() + .withExternal() + .source("S1") + .target("S2") + .and() + .withExternal() + .source("S2") + .target("S3"); + } + + @Bean + public HeaderTestAction headerTestAction1() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction headerTestAction2() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction headerTestAction3() { + return new HeaderTestAction(); + } + } + + @Configuration + @EnableStateMachine + static class Config4 extends StateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial("SI") + .state("S1", headerTestAction1(), null) + .state("S2", headerTestAction2(), null) + .state("S3", headerTestAction3(), null); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions + .withExternal() + .source("SI") + .target("S1") + .event("E1") + .and() + .withExternal() + .source("S1") + .target("S2") + .and() + .withExternal() + .source("S2") + .target("S3"); + } + + @Bean + public HeaderTestAction headerTestAction1() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction headerTestAction2() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction headerTestAction3() { + return new HeaderTestAction(); + } + + @Bean(name = StateMachineSystemConstants.TASK_EXECUTOR_BEAN_NAME) + public TaskExecutor taskExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(3); + return executor; + } + } + private static class HeaderTestAction implements Action { String testHeader = null; diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionTests.java index d7c0bc8b..db92e4ba 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 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,13 +28,19 @@ import java.util.EnumSet; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; 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; @@ -52,6 +58,8 @@ import org.springframework.statemachine.state.State; */ public class TransitionTests extends AbstractStateMachineTests { + private final static Log log = LogFactory.getLog(TransitionTests.class); + @Override protected AnnotationConfigApplicationContext buildContext() { return new AnnotationConfigApplicationContext(); @@ -229,6 +237,126 @@ public class TransitionTests extends AbstractStateMachineTests { assertThat(machine.getState().getIds(), contains(TestStates2.BUSY, TestStates2.PAUSED, TestStates2.PAUSED2)); } + @SuppressWarnings("unchecked") + @Test + public void testAnonymousTransitionInSubmachine() throws InterruptedException { + context.register(BaseConfig.class, Config9.class); + context.refresh(); + + assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); + ObjectStateMachine machine = + context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + HeaderTestAction testAction1 = context.getBean("testAction1", HeaderTestAction.class); + HeaderTestAction testAction2 = context.getBean("testAction2", HeaderTestAction.class); + + + TestListener listener = new TestListener(); + machine.addStateListener(listener); + + machine.start(); + assertThat(machine.getState().getIds(), contains(TestStates.S1)); + + listener.reset(4); + machine.sendEvent(MessageBuilder.withPayload(TestEvents.E1).setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(4)); + assertThat(machine.getState().getIds(), contains(TestStates.S2, TestStates.S212)); + + assertThat(testAction1.testHeader, is("testValue")); + assertThat(testAction2.testHeader, is("testValue")); + } + + @SuppressWarnings("unchecked") + @Test + public void testAnonymousTransitionInSubmachineWithThreading() throws InterruptedException { + context.register(Config9.class, ExecutorConfig.class); + context.refresh(); + + assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); + ObjectStateMachine machine = + context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + HeaderTestAction testAction1 = context.getBean("testAction1", HeaderTestAction.class); + HeaderTestAction testAction2 = context.getBean("testAction2", HeaderTestAction.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(4); + machine.sendEvent(MessageBuilder.withPayload(TestEvents.E1).setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(4)); + assertThat(machine.getState().getIds(), contains(TestStates.S2, TestStates.S212)); + + assertThat(testAction1.testHeader, is("testValue")); + assertThat(testAction2.testHeader, is("testValue")); + } + + @SuppressWarnings("unchecked") + @Test + public void testAnonymousTransitionInSubmachineWithExitWithThreading1() throws InterruptedException { + context.register(Config10.class, ExecutorConfig.class); + context.refresh(); + + assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); + ObjectStateMachine machine = + context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + HeaderTestAction testAction1 = context.getBean("testAction1", HeaderTestAction.class); + HeaderTestAction testAction2 = context.getBean("testAction2", HeaderTestAction.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(5); + machine.sendEvent(MessageBuilder.withPayload(TestEvents.E1).setHeader("testHeader", "testValue").build()); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(5)); + assertThat(machine.getState().getIds(), contains(TestStates.S1)); + + assertThat(testAction1.testHeader, is("testValue")); + assertThat(testAction2.testHeader, is("testValue")); + } + + @SuppressWarnings("unchecked") + @Test + public void testAnonymousTransitionInSubmachineWithExitWithThreading2() throws InterruptedException { + context.register(Config11.class, ExecutorConfig.class); + context.refresh(); + + assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); + ObjectStateMachine machine = + context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); + HeaderTestAction testAction1 = context.getBean("testAction1", HeaderTestAction.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("testHeader", "testValue").build()); + assertThat(testAction1.latch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.s20Latch.getCount(), is(1L)); + + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(3)); + assertThat(machine.getState().getIds(), contains(TestStates.S1)); + + assertThat(testAction1.testHeader, is("testValue")); + } + @Configuration @EnableStateMachine public static class Config1 extends EnumStateMachineConfigurerAdapter { @@ -555,11 +683,180 @@ public class TransitionTests extends AbstractStateMachineTests { } + @Configuration + @EnableStateMachine + public static class Config9 extends EnumStateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial(TestStates.S1) + .state(TestStates.S2) + .and() + .withStates() + .parent(TestStates.S2) + .initial(TestStates.S20) + .stateEntry(TestStates.S211, testAction1()) + .stateEntry(TestStates.S212, testAction2()); + } + + @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() + .withExternal() + .source(TestStates.S211) + .target(TestStates.S212); + } + + @Bean + public HeaderTestAction testAction1() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction testAction2() { + return new HeaderTestAction(); + } + } + + @Configuration + @EnableStateMachine + public static class Config10 extends EnumStateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial(TestStates.S1) + .state(TestStates.S2) + .and() + .withStates() + .parent(TestStates.S2) + .initial(TestStates.S20) + .stateEntry(TestStates.S211, testAction1()) + .stateEntry(TestStates.S212, testAction2()) + .exit(TestStates.SF); + } + + @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() + .withExternal() + .source(TestStates.S211) + .target(TestStates.S212) + .and() + .withExternal() + .source(TestStates.S212) + .target(TestStates.SF) + .and() + .withExit() + .source(TestStates.SF) + .target(TestStates.S1); + } + + @Bean + public HeaderTestAction testAction1() { + return new HeaderTestAction(); + } + + @Bean + public HeaderTestAction testAction2() { + return new HeaderTestAction(); + } + } + + @Configuration + @EnableStateMachine + public static class Config11 extends EnumStateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial(TestStates.S1) + .state(TestStates.S2) + .and() + .withStates() + .parent(TestStates.S2) + .initial(TestStates.S20) + .stateEntry(TestStates.S20, testAction1()) + .exit(TestStates.SF); + } + + @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.SF) + .and() + .withExit() + .source(TestStates.SF) + .target(TestStates.S1); + } + + @Bean + public HeaderTestAction testAction1() { + return new HeaderTestAction(); + } + } + + private static class HeaderTestAction implements Action { + + volatile CountDownLatch latch = new CountDownLatch(1); + String testHeader = null; + + @Override + public void execute(StateContext context) { + log.info("XXX11"); + try { + Thread.sleep(500); + } catch (InterruptedException e) { + } + testHeader = context.getMessageHeaders().get("testHeader", String.class); + log.info("XXX12"); + latch.countDown(); + } + + + } + static class TestListener extends StateMachineListenerAdapter { + volatile CountDownLatch stateMachineStartedLatch = new CountDownLatch(1); volatile CountDownLatch stateChangedLatch = new CountDownLatch(1); + volatile CountDownLatch s20Latch = new CountDownLatch(1); volatile int stateChangedCount = 0; + @Override + public void stateMachineStarted(StateMachine stateMachine) { + stateMachineStartedLatch.countDown(); + } + @Override public void stateChanged(State from, State to) { stateChangedCount++; @@ -571,6 +868,14 @@ public class TransitionTests extends AbstractStateMachineTests { stateChangedCount = 0; } + @Override + public void stateExited(State state) { + if (state.getId() == TestStates.S20) { + log.info("XXX22"); + s20Latch.countDown(); + } + } + } static class TestListener2 extends StateMachineListenerAdapter { @@ -605,4 +910,15 @@ public class TransitionTests extends AbstractStateMachineTests { } + @Configuration + static class ExecutorConfig { + + @Bean(name=StateMachineSystemConstants.TASK_EXECUTOR_BEAN_NAME) + public TaskExecutor taskExecutor() { + ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); + taskExecutor.setCorePoolSize(3); + taskExecutor.setMaxPoolSize(3); + return taskExecutor; + } + } }