Spring/spring-statemachine
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix threading issues

Browse Source 
- 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
This commit is contained in:
Janne Valkealahti
2017-11-26 13:54:11 +00:00
parent 6ba231af2e
commit 2a53b4eee5
5 changed files with 754 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java
View File

@@ -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<S, E> extends StateMachineObjectSuppo
public State<S,E> 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, E> s = lastState;
if (s != null && isComplete()) {
return s;
} else {
return currentState;
}
}
@@ -286,6 +286,19 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
@Override
public void transit(Transition<S, E> t, StateContext<S, E> ctx, Message<E> 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<S, E> submachine = ((AbstractState<S, E>)currentState).getSubmachine();
Lock lock = ((AbstractStateMachine<S, E>)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<S, E> extends StateMachineObjectSuppo
}
}
protected StateMachineExecutor<S, E> getStateMachineExecutor() {
return stateMachineExecutor;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
// last change to set factory because this maybe be called per

45
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java
View File

@@ -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<S, E> extends LifecycleObjectSupport im
private volatile Message<E> queuedMessage = null;
private final ReentrantLock lock = new ReentrantLock();
/**
* Instantiates a new default state machine executor.
*
@@ -183,6 +187,11 @@ public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport im
interceptors.add(interceptor);
}
@Override
public Lock getLock() {
return lock;
}
private final Set<Transition<S, E>> joinSyncTransitions = new HashSet<>();
private final Set<State<S, E>> joinSyncStates = new HashSet<>();
@@ -274,29 +283,37 @@ public class DefaultStateMachineExecutor<S, E> 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)) {

11
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java
View File

@@ -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<S, E> {
*/
void addStateMachineInterceptor(StateMachineInterceptor<S, E> interceptor);
/**
* Gets the execution lock.
*
* @return the execution lock
*/
Lock getLock();
/**
* Callback interface when executor wants to handle transit.
*/

374
spring-statemachine-core/src/test/java/org/springframework/statemachine/EventHeaderTests.java
View File

@@ -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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> {
@@ -248,6 +443,183 @@ public class EventHeaderTests extends AbstractStateMachineTests {
}
}
@Configuration
@EnableStateMachine
static class Config2 extends StateMachineConfigurerAdapter<String, String> {
@Override
public void configure(StateMachineStateConfigurer<String, String> 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<String, String> 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<String, String> {
@Override
public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
states
.withStates()
.initial("SI")
.state("S1", headerTestAction1(), null)
.state("S2", headerTestAction2(), null)
.state("S3", headerTestAction3(), null);
}
@Override
public void configure(StateMachineTransitionConfigurer<String, String> 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<String, String> {
@Override
public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
states
.withStates()
.initial("SI")
.state("S1", headerTestAction1(), null)
.state("S2", headerTestAction2(), null)
.state("S3", headerTestAction3(), null);
}
@Override
public void configure(StateMachineTransitionConfigurer<String, String> 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, String> {
String testHeader = null;

318
spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionTests.java
View File

@@ -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<TestStates,TestEvents> 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<TestStates,TestEvents> 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<TestStates,TestEvents> 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<TestStates,TestEvents> 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<TestStates, TestEvents> {
@@ -555,11 +683,180 @@ public class TransitionTests extends AbstractStateMachineTests {
}
@Configuration
@EnableStateMachine
public static class Config9 extends EnumStateMachineConfigurerAdapter<TestStates, TestEvents> {
@Override
public void configure(StateMachineStateConfigurer<TestStates, TestEvents> 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<TestStates, TestEvents> 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<TestStates, TestEvents> {
@Override
public void configure(StateMachineStateConfigurer<TestStates, TestEvents> 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<TestStates, TestEvents> 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<TestStates, TestEvents> {
@Override
public void configure(StateMachineStateConfigurer<TestStates, TestEvents> 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<TestStates, TestEvents> 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<TestStates, TestEvents> {
volatile CountDownLatch latch = new CountDownLatch(1);
String testHeader = null;
@Override
public void execute(StateContext<TestStates, TestEvents> 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<TestStates, TestEvents> {
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<TestStates, TestEvents> stateMachine) {
stateMachineStartedLatch.countDown();
}
@Override
public void stateChanged(State<TestStates, TestEvents> from, State<TestStates, TestEvents> to) {
stateChangedCount++;
@@ -571,6 +868,14 @@ public class TransitionTests extends AbstractStateMachineTests {
stateChangedCount = 0;
}
@Override
public void stateExited(State<TestStates, TestEvents> state) {
if (state.getId() == TestStates.S20) {
log.info("XXX22");
s20Latch.countDown();
}
}
}
static class TestListener2 extends StateMachineListenerAdapter<TestStates2, TestEvents2> {
@@ -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;
}
}
}