From 2a53b4eee5678683c00bc235256bea7cf2922877 Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Sun, 26 Nov 2017 13:54:11 +0000 Subject: [PATCH] Fix threading issues - In AbstractStateMachine and DefaultStateMachineExecutor do a little more synchronisation so that submachine is allowed to do its run-to-completion before parent machine can do its own transitions. This should fix use cases, when submachine starts, does its transitions and actions, parent wont try to do its own triggerless transtions. - Adding some new tests and fixing one other sync issue with end state. - Relates to #442 --- .../support/AbstractStateMachine.java | 29 +- .../support/DefaultStateMachineExecutor.java | 45 ++- .../support/StateMachineExecutor.java | 11 +- .../statemachine/EventHeaderTests.java | 374 +++++++++++++++++- .../transition/TransitionTests.java | 318 ++++++++++++++- 5 files changed, 754 insertions(+), 23 deletions(-) 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; + } + } }