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

Better support for parallel regions

Browse Source 
- Generally fixes #68
- First attempt to externalize event execution from state machine
  into its own class backed by an interface. This relates to #7
- Change of various places to have better support if execution
  is done in threads.
- Not yet a central place where concurrency can be defines, thus
  currently rely on global taskExecutor bean when can be overridded
  from a default which is SyncTaskExecutor. Futher work for
  that in separate tickets.
- Change tasks sample to use a thread pool.
- Change of concept how initial state/transition is handled, no
  longer handled manually in lifecycle method, thus giving a change
  for initial transition to execute its actions in a multiple threads.
This commit is contained in:
Janne Valkealahti
2015-05-24 17:38:41 +01:00
parent 517bf69364
commit 2b3db24345
18 changed files with 866 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-statemachine-core/src/main/java/org/springframework/statemachine/config/EnumStateMachineFactory.java
View File

@@ -46,9 +46,9 @@ import org.springframework.statemachine.state.RegionState;
import org.springframework.statemachine.state.State;
import org.springframework.statemachine.state.StateMachineState;
import org.springframework.statemachine.support.DefaultExtendedState;
import org.springframework.statemachine.support.StateMachineFunction;
import org.springframework.statemachine.support.LifecycleObjectSupport;
import org.springframework.statemachine.support.StateMachineAccess;
import org.springframework.statemachine.support.StateMachineFunction;
import org.springframework.statemachine.support.tree.Tree;
import org.springframework.statemachine.support.tree.Tree.Node;
import org.springframework.statemachine.support.tree.TreeTraverser;
@@ -161,8 +161,9 @@ public class EnumStateMachineFactory<S extends Enum<S>, E extends Enum<E>> exten
// TODO: don't like that we create a last machine here
Collection<State<S, E>> states = new ArrayList<State<S, E>>();
states.add(rstate);
Transition<S, E> initialTransition = new InitialTransition<S, E>(rstate);
EnumStateMachine<S, E> m = new EnumStateMachine<S, E>(states, new ArrayList<Transition<S, E>>(), rstate,
null, null, defaultExtendedState);
initialTransition, null, defaultExtendedState);
if (contextEvents != null) {
m.setContextEventsEnabled(contextEvents);
}
@@ -460,22 +461,17 @@ public class EnumStateMachineFactory<S extends Enum<S>, E extends Enum<E>> exten
}
}
// TODO: should make a proper transition
Transition<S, E> initialTransition = null;
if (initialAction != null) {
initialTransition = new InitialTransition<S, E>(initialState, initialAction);
}
Transition<S, E> initialTransition = new InitialTransition<S, E>(initialState, initialAction);
EnumStateMachine<S, E> machine = new EnumStateMachine<S, E>(states, transitions, initialState,
initialTransition, null, defaultExtendedState);
machine.setHistoryState(historyState);
if (contextEvents != null) {
machine.setContextEventsEnabled(contextEvents);
}
machine.afterPropertiesSet();
if (beanFactory != null) {
machine.setBeanFactory(beanFactory);
}
machine.afterPropertiesSet();
return machine;
}

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

@@ -21,23 +21,16 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.core.OrderComparator;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.task.TaskExecutor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
@@ -58,12 +51,12 @@ import org.springframework.statemachine.state.PseudoStateContext;
import org.springframework.statemachine.state.PseudoStateKind;
import org.springframework.statemachine.state.PseudoStateListener;
import org.springframework.statemachine.state.State;
import org.springframework.statemachine.support.StateMachineExecutor.StateMachineExecutorTransit;
import org.springframework.statemachine.transition.InitialTransition;
import org.springframework.statemachine.transition.Transition;
import org.springframework.statemachine.transition.TransitionKind;
import org.springframework.statemachine.trigger.DefaultTriggerContext;
import org.springframework.statemachine.trigger.TimerTrigger;
import org.springframework.statemachine.trigger.Trigger;
import org.springframework.statemachine.trigger.TriggerListener;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -92,30 +85,22 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
private final ExtendedState extendedState;
private final Queue<Message<E>> eventQueue = new ConcurrentLinkedQueue<Message<E>>();
private final LinkedList<Message<E>> deferList = new LinkedList<Message<E>>();
private volatile State<S,E> currentState;
private volatile PseudoState<S, E> history;
private volatile Runnable task;
private final AtomicBoolean requestTask = new AtomicBoolean(false);
private final Map<String, StateMachineOnTransitionHandler<S, E>> handlers = new HashMap<String, StateMachineOnTransitionHandler<S,E>>();
private volatile boolean handlersInitialized;
private final Queue<TriggerQueueItem> triggerQueue = new ConcurrentLinkedQueue<TriggerQueueItem>();
private final Map<Trigger<S, E>, Transition<S,E>> triggerToTransitionMap = new HashMap<Trigger<S,E>, Transition<S,E>>();
private final List<Transition<S, E>> triggerlessTransitions = new ArrayList<Transition<S,E>>();
private StateMachine<S, E> relay;
private StateMachineExecutor<S, E> stateMachineExecutor;
/**
* Instantiates a new abstract state machine.
*
@@ -156,9 +141,13 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
this.states = states;
this.transitions = transitions;
this.initialState = initialState;
this.initialTransition = initialTransition;
this.initialEvent = initialEvent;
this.extendedState = extendedState != null ? extendedState : new DefaultExtendedState();
if (initialTransition == null) {
this.initialTransition = new InitialTransition<S, E>(initialState);
} else {
this.initialTransition = initialTransition;
}
}
@Override
@@ -186,7 +175,7 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
return false;
}
boolean accepted = acceptEvent(event);
scheduleEventQueueProcessing();
stateMachineExecutor.execute();
return accepted;
}
@@ -202,6 +191,8 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
Assert.state(initialState.getPseudoState() != null
&& initialState.getPseudoState().getKind() == PseudoStateKind.INITIAL,
"Initial state's pseudostate kind must be INITIAL");
// process given transitions
for (Transition<S, E> transition : transitions) {
Trigger<S, E> trigger = transition.getTrigger();
if (trigger != null) {
@@ -211,6 +202,7 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
triggerlessTransitions.add(transition);
}
}
for (State<S, E> state : states) {
if (state.isSubmachineState()) {
StateMachine<S, E> submachine = ((AbstractState<S, E>)state).getSubmachine();
@@ -227,6 +219,32 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
history = state.getPseudoState();
}
}
DefaultStateMachineExecutor<S, E> executor = new DefaultStateMachineExecutor<S, E>(this, getRelayStateMachine(), extendedState,
transitions, triggerToTransitionMap, triggerlessTransitions, initialTransition, initialEvent);
if (getBeanFactory() != null) {
executor.setBeanFactory(getBeanFactory());
} else if (getTaskExecutor() != null){
executor.setTaskExecutor(getTaskExecutor());
}
executor.afterPropertiesSet();
executor.setStateMachineExecutorTransit(new StateMachineExecutorTransit<S, E>() {
@Override
public void transit(Transition<S, E> t, StateContext<S, E> stateContext, Message<E> queuedMessage) {
notifyTransitionStart(t);
callHandlers(t.getSource(), t.getTarget(), queuedMessage);
if (t.getKind() == TransitionKind.INITIAL) {
switchToState(t.getTarget(), queuedMessage, null, getRelayStateMachine());
notifyStateMachineStarted(getRelayStateMachine());
} else if (t.getKind() != TransitionKind.INTERNAL) {
switchToState(t.getTarget(), queuedMessage, t, getRelayStateMachine());
}
notifyTransition(t);
notifyTransitionEnd(t);
}
});
stateMachineExecutor = executor;
}
@Override
@@ -235,21 +253,15 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
if (currentState != null) {
return;
}
registerTriggerListener();
registerPseudoStateListener();
switchToState(initialState, initialEvent, null, getRelayStateMachine());
// TODO: it is a bit off to call handlers after switchToState for initial state
callHandlers(null, initialState, initialEvent);
// TODO: for now execute outside of switchToState
if (initialTransition != null) {
StateContext<S, E> stateContext = buildStateContext(initialEvent, initialTransition, getRelayStateMachine());
initialTransition.transit(stateContext);
}
notifyStateMachineStarted(this);
// start fires first execution which should execute initial transition
stateMachineExecutor.start();
}
@Override
protected void doStop() {
stateMachineExecutor.stop();
notifyStateMachineStopped(this);
currentState = null;
}
@@ -313,6 +325,12 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
this.relay = stateMachine;
}
@Override
protected void stateChangedInRelay() {
// TODO: temp tweak, see super
stateMachineExecutor.execute();
}
private StateMachine<S, E> getRelayStateMachine() {
return relay != null ? relay : this;
}
@@ -352,7 +370,7 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
if (StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) {
if (trigger != null && trigger.evaluate(new DefaultTriggerContext<S, E>(message.getPayload()))) {
triggerQueue.add(new TriggerQueueItem(trigger, message));
stateMachineExecutor.queueTrigger(trigger, message);
return true;
} else if (source.getDeferredEvents() != null && source.getDeferredEvents().contains(message.getPayload())) {
defer = message;
@@ -361,7 +379,7 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
}
if (defer != null) {
log.info("Deferring event " + defer);
deferList.addLast(defer);
stateMachineExecutor.queueDeferredEvent(defer);
return true;
}
@@ -386,7 +404,7 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
setCurrentState(state, message, transition, true, stateMachine);
}
scheduleEventQueueProcessing();
stateMachineExecutor.execute();
if (isComplete()) {
stop();
}
@@ -436,6 +454,7 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
}
void setCurrentState(State<S, E> state, Message<E> message, Transition<S, E> transition, boolean exit, StateMachine<S, E> stateMachine) {
State<S, E> findDeep = findDeepParent(state);
boolean isTargetSubOf = false;
if (transition != null) {
@@ -597,157 +616,6 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
state.entry(stateContext);
}
private void processEventQueue() {
log.debug("Process event queue");
Message<E> queuedEvent = null;
while ((queuedEvent = eventQueue.poll()) != null) {
Message<E> defer = null;
for (Transition<S,E> transition : transitions) {
State<S,E> source = transition.getSource();
Trigger<S, E> trigger = transition.getTrigger();
if (StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) {
if (trigger != null && trigger.evaluate(new DefaultTriggerContext<S, E>(queuedEvent.getPayload()))) {
triggerQueue.add(new TriggerQueueItem(trigger, queuedEvent));
} else if (source.getDeferredEvents() != null && source.getDeferredEvents().contains(queuedEvent.getPayload())) {
defer = queuedEvent;
}
}
}
if (defer != null) {
log.info("Deferring event " + defer);
deferList.addLast(defer);
}
}
}
private boolean processDeferList() {
log.debug("Process defer list");
boolean triggered = false;
ListIterator<Message<E>> iterator = deferList.listIterator();
while (iterator.hasNext()) {
Message<E> event = iterator.next();
for (Transition<S,E> transition : transitions) {
State<S,E> source = transition.getSource();
Trigger<S, E> trigger = transition.getTrigger();
if (source.equals(currentState)) {
if (trigger != null && trigger.evaluate(new DefaultTriggerContext<S, E>(event.getPayload()))) {
triggerQueue.add(new TriggerQueueItem(trigger, event));
iterator.remove();
triggered = true;
}
}
}
}
return triggered;
}
private void scheduleEventQueueProcessing() {
TaskExecutor executor = getTaskExecutor();
if (executor == null) {
return;
}
if (task == null) {
task = new Runnable() {
@Override
public void run() {
processEventQueue();
processTriggerQueue();
while (processDeferList()) {
processTriggerQueue();
}
task = null;
if (requestTask.getAndSet(false)) {
scheduleEventQueueProcessing();
}
}
};
executor.execute(task);
} else {
requestTask.set(true);
}
}
private void processTriggerQueue() {
log.debug("Process trigger queue");
TriggerQueueItem queueItem = null;
while ((queueItem = triggerQueue.poll()) != null) {
if (currentState == null) {
continue;
}
Message<E> queuedMessage = queueItem.message;
E event = queuedMessage != null ? queuedMessage.getPayload() : null;
// need all transitions trigger could match, event trigger may match multiple
// need to go up from substates and ask if trigger transit, if not check super
ArrayList<Transition<S, E>> trans = new ArrayList<Transition<S,E>>();
if (event != null) {
ArrayList<S> ids = new ArrayList<S>(currentState.getIds());
Collections.reverse(ids);
for (S id : ids) {
for (Entry<Trigger<S, E>, Transition<S, E>> e : triggerToTransitionMap.entrySet()) {
Trigger<S, E> tri = e.getKey();
E ee = tri.getEvent();
Transition<S, E> tra = e.getValue();
if (event == ee) {
if (tra.getSource().getId() == id && !trans.contains(tra)) {
trans.add(tra);
continue;
}
}
}
}
}
// most likely timer
if (trans.isEmpty()) {
trans.add(triggerToTransitionMap.get(queueItem.trigger));
}
// go through candidates and transit max one
handleTriggerTrans(trans, queuedMessage);
}
if (currentState != null) {
// handle triggerless transitions
handleTriggerTrans(triggerlessTransitions, null);
}
}
private void handleTriggerTrans(List<Transition<S, E>> trans, Message<E> queuedMessage) {
for (Transition<S, E> t : trans) {
StateContext<S, E> stateContext = buildStateContext(queuedMessage, t, getRelayStateMachine());
if (t == null) {
continue;
}
State<S,E> source = t.getSource();
if (source == null) {
continue;
}
if (!StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) {
continue;
}
boolean transit = t.transit(stateContext);
if (transit) {
// TODO: should change trasition api so that we can ask
// if transition will transit so that we can post
// accurate notifyTransitionStart
notifyTransitionStart(t);
callHandlers(t.getSource(), t.getTarget(), queuedMessage);
if (t.getKind() != TransitionKind.INTERNAL) {
switchToState(t.getTarget(), queuedMessage, t, getRelayStateMachine());
}
notifyTransition(t);
notifyTransitionEnd(t);
break;
}
}
}
private void callHandlers(State<S,E> sourceState, State<S,E> targetState, Message<E> message) {
StateContext<S, E> stateContext = buildStateContext(message, null, getRelayStateMachine());
getStateMachineHandlerResults(getStateMachineHandlers(sourceState, targetState), stateContext);
@@ -847,31 +715,4 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
return handle;
}
private void registerTriggerListener() {
for (final Trigger<S, E> trigger : triggerToTransitionMap.keySet()) {
if (trigger instanceof TimerTrigger) {
((TimerTrigger<?, ?>)trigger).addTriggerListener(new TriggerListener() {
@Override
public void triggered() {
log.debug("TimedTrigger triggered " + trigger);
triggerQueue.add(new TriggerQueueItem(trigger, null));
scheduleEventQueueProcessing();
}
});
}
if (trigger instanceof Lifecycle) {
((Lifecycle)trigger).start();
}
}
}
private class TriggerQueueItem {
Trigger<S, E> trigger;
Message<E> message;
public TriggerQueueItem(Trigger<S, E> trigger, Message<E> message) {
this.trigger = trigger;
this.message = message;
}
}
}

347
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java Normal file
View File

@@ -0,0 +1,347 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.statemachine.support;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.TaskExecutor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.statemachine.ExtendedState;
import org.springframework.statemachine.StateContext;
import org.springframework.statemachine.StateMachine;
import org.springframework.statemachine.state.State;
import org.springframework.statemachine.transition.Transition;
import org.springframework.statemachine.trigger.DefaultTriggerContext;
import org.springframework.statemachine.trigger.TimerTrigger;
import org.springframework.statemachine.trigger.Trigger;
import org.springframework.statemachine.trigger.TriggerListener;
/**
* Default implementation of a {@link StateMachineExecutor}.
*
* @author Janne Valkealahti
*
* @param <S> the type of state
* @param <E> the type of event
*/
public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport implements StateMachineExecutor<S, E> {
private static final Log log = LogFactory.getLog(DefaultStateMachineExecutor.class);
private final StateMachine<S, E> stateMachine;
private final StateMachine<S, E> relayStateMachine;
private final ExtendedState extendedState;
private final Queue<Message<E>> eventQueue = new ConcurrentLinkedQueue<Message<E>>();
private final LinkedList<Message<E>> deferList = new LinkedList<Message<E>>();
private final Queue<TriggerQueueItem> triggerQueue = new ConcurrentLinkedQueue<TriggerQueueItem>();
private final Collection<Transition<S,E>> transitions;
private final AtomicBoolean requestTask = new AtomicBoolean(false);
private final Map<Trigger<S, E>, Transition<S,E>> triggerToTransitionMap;
private final List<Transition<S, E>> triggerlessTransitions;
private final Transition<S, E> initialTransition;
private final Message<E> initialEvent;
private final AtomicBoolean initialHandled = new AtomicBoolean(false);
private volatile Runnable task;
private StateMachineExecutorTransit<S, E> stateMachineExecutorTransit;
/**
* Instantiates a new default state machine executor.
*
* @param stateMachine the state machine
* @param relayStateMachine the relay state machine
* @param extendedState the extended state
* @param transitions the transitions
* @param triggerToTransitionMap the trigger to transition map
* @param triggerlessTransitions the triggerless transitions
*/
public DefaultStateMachineExecutor(StateMachine<S, E> stateMachine, StateMachine<S, E> relayStateMachine,
ExtendedState extendedState, Collection<Transition<S, E>> transitions, Map<Trigger<S, E>, Transition<S, E>> triggerToTransitionMap,
List<Transition<S, E>> triggerlessTransitions, Transition<S, E> initialTransition, Message<E> initialEvent) {
this.stateMachine = stateMachine;
this.relayStateMachine = relayStateMachine;
this.extendedState = extendedState;
this.triggerToTransitionMap = triggerToTransitionMap;
this.triggerlessTransitions = triggerlessTransitions;
this.transitions = transitions;
this.initialTransition = initialTransition;
this.initialEvent = initialEvent;
registerTriggerListener();
}
@Override
public void queueEvent(Message<E> message) {
eventQueue.add(message);
}
@Override
public void queueTrigger(Trigger<S, E> trigger, Message<E> message) {
triggerQueue.add(new TriggerQueueItem(trigger, message));
}
@Override
public void queueDeferredEvent(Message<E> message) {
deferList.addLast(message);
}
@Override
public void execute() {
scheduleEventQueueProcessing();
}
@Override
public void setStateMachineExecutorTransit(StateMachineExecutorTransit<S, E> stateMachineExecutorTransit) {
this.stateMachineExecutorTransit = stateMachineExecutorTransit;
}
@Override
protected void doStart() {
super.doStart();
execute();
}
@Override
protected void doStop() {
super.doStop();
initialHandled.set(false);
}
private void handleTriggerTrans(List<Transition<S, E>> trans, Message<E> queuedMessage) {
for (Transition<S, E> t : trans) {
if (t == null) {
continue;
}
State<S,E> source = t.getSource();
if (source == null) {
continue;
}
State<S,E> currentState = stateMachine.getState();
if (!StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) {
continue;
}
StateContext<S, E> stateContext = buildStateContext(queuedMessage, t, relayStateMachine);
boolean transit = t.transit(stateContext);
if (transit) {
stateMachineExecutorTransit.transit(t, stateContext, queuedMessage);
break;
}
}
}
private void handleInitialTrans(Transition<S, E> tran, Message<E> queuedMessage) {
StateContext<S, E> stateContext = buildStateContext(queuedMessage, tran, relayStateMachine);
tran.transit(stateContext);
stateMachineExecutorTransit.transit(tran, stateContext, queuedMessage);
}
private void scheduleEventQueueProcessing() {
TaskExecutor executor = getTaskExecutor();
if (executor == null) {
return;
}
if (task == null) {
task = new Runnable() {
@Override
public void run() {
processEventQueue();
processTriggerQueue();
while (processDeferList()) {
processTriggerQueue();
}
task = null;
if (requestTask.getAndSet(false)) {
scheduleEventQueueProcessing();
}
}
};
executor.execute(task);
} else {
requestTask.set(true);
}
}
private void processEventQueue() {
log.debug("Process event queue");
Message<E> queuedEvent = null;
State<S,E> currentState = stateMachine.getState();
while ((queuedEvent = eventQueue.poll()) != null) {
Message<E> defer = null;
for (Transition<S, E> transition : transitions) {
State<S, E> source = transition.getSource();
Trigger<S, E> trigger = transition.getTrigger();
if (StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) {
if (trigger != null && trigger.evaluate(new DefaultTriggerContext<S, E>(queuedEvent.getPayload()))) {
triggerQueue.add(new TriggerQueueItem(trigger, queuedEvent));
} else if (source.getDeferredEvents() != null
&& source.getDeferredEvents().contains(queuedEvent.getPayload())) {
defer = queuedEvent;
}
}
}
if (defer != null) {
log.info("Deferring event " + defer);
deferList.addLast(defer);
}
}
}
private void processTriggerQueue() {
if (!isRunning()) {
return;
}
if (!initialHandled.getAndSet(true)) {
ArrayList<Transition<S, E>> trans = new ArrayList<Transition<S, E>>();
trans.add(initialTransition);
handleInitialTrans(initialTransition, initialEvent);
return;
}
log.debug("Process trigger queue");
TriggerQueueItem queueItem = null;
while ((queueItem = triggerQueue.poll()) != null) {
State<S,E> currentState = stateMachine.getState();
if (currentState == null) {
continue;
}
Message<E> queuedMessage = queueItem.message;
E event = queuedMessage != null ? queuedMessage.getPayload() : null;
// need all transitions trigger could match, event trigger may match
// multiple
// need to go up from substates and ask if trigger transit, if not
// check super
ArrayList<Transition<S, E>> trans = new ArrayList<Transition<S, E>>();
if (event != null) {
ArrayList<S> ids = new ArrayList<S>(currentState.getIds());
Collections.reverse(ids);
for (S id : ids) {
for (Entry<Trigger<S, E>, Transition<S, E>> e : triggerToTransitionMap.entrySet()) {
Trigger<S, E> tri = e.getKey();
E ee = tri.getEvent();
Transition<S, E> tra = e.getValue();
if (event == ee) {
if (tra.getSource().getId() == id && !trans.contains(tra)) {
trans.add(tra);
continue;
}
}
}
}
}
// most likely timer
if (trans.isEmpty()) {
trans.add(triggerToTransitionMap.get(queueItem.trigger));
}
// go through candidates and transit max one
handleTriggerTrans(trans, queuedMessage);
}
if (stateMachine.getState() != null) {
// handle triggerless transitions
handleTriggerTrans(triggerlessTransitions, null);
}
}
private boolean processDeferList() {
log.debug("Process defer list");
boolean triggered = false;
ListIterator<Message<E>> iterator = deferList.listIterator();
State<S,E> currentState = stateMachine.getState();
while (iterator.hasNext()) {
Message<E> event = iterator.next();
for (Transition<S, E> transition : transitions) {
State<S, E> source = transition.getSource();
Trigger<S, E> trigger = transition.getTrigger();
if (source.equals(currentState)) {
if (trigger != null && trigger.evaluate(new DefaultTriggerContext<S, E>(event.getPayload()))) {
triggerQueue.add(new TriggerQueueItem(trigger, event));
iterator.remove();
triggered = true;
}
}
}
}
return triggered;
}
private StateContext<S, E> buildStateContext(Message<E> message, Transition<S,E> transition, StateMachine<S, E> stateMachine) {
E event = message != null ? message.getPayload() : null;
MessageHeaders messageHeaders = message != null ? message.getHeaders() : new MessageHeaders(
new HashMap<String, Object>());
return new DefaultStateContext<S, E>(event, messageHeaders, extendedState, transition, stateMachine);
}
private void registerTriggerListener() {
for (final Trigger<S, E> trigger : triggerToTransitionMap.keySet()) {
if (trigger instanceof TimerTrigger) {
((TimerTrigger<?, ?>) trigger).addTriggerListener(new TriggerListener() {
@Override
public void triggered() {
log.debug("TimedTrigger triggered " + trigger);
triggerQueue.add(new TriggerQueueItem(trigger, null));
scheduleEventQueueProcessing();
}
});
}
if (trigger instanceof Lifecycle) {
((Lifecycle) trigger).start();
}
}
}
private class TriggerQueueItem {
Trigger<S, E> trigger;
Message<E> message;
public TriggerQueueItem(Trigger<S, E> trigger, Message<E> message) {
this.trigger = trigger;
this.message = message;
}
}
}

98
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java Normal file
View File

@@ -0,0 +1,98 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.statemachine.support;
import org.springframework.messaging.Message;
import org.springframework.statemachine.StateContext;
import org.springframework.statemachine.StateMachine;
import org.springframework.statemachine.transition.Transition;
import org.springframework.statemachine.trigger.Trigger;
/**
* Interface for a {@link StateMachine} event executor.
*
* @author Janne Valkealahti
*
* @param <S> the type of state
* @param <E> the type of event
*/
public interface StateMachineExecutor<S, E> {
/**
* Queue event.
*
* @param message the message
*/
void queueEvent(Message<E> message);
/**
* Queue trigger.
*
* @param trigger the trigger
* @param message the message
*/
void queueTrigger(Trigger<S, E> trigger, Message<E> message);
/**
* Queue deferred event.
*
* @param message the message
*/
void queueDeferredEvent(Message<E> message);
/**
* Execute {@code StateMachineExecutor} logic.
*/
void execute();
/**
* Start executor.
*
* @see LifecycleObjectSupport#start()
*/
void start();
/**
* Stop executor.
*
* @see LifecycleObjectSupport#stop()
*/
void stop();
/**
* Sets the state machine executor transit.
*
* @param stateMachineExecutorTransit the state machine executor transit
*/
void setStateMachineExecutorTransit(StateMachineExecutorTransit<S, E> stateMachineExecutorTransit);
/**
* Callback interface when executor wants to handle transit.
*/
public interface StateMachineExecutorTransit<S, E> {
/**
* Called when executor wants to do a transit.
*
* @param transition the transition
* @param stateContext the state context
* @param message the message
*/
void transit(Transition<S, E> transition, StateContext<S, E> stateContext, Message<E> message);
}
}

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

@@ -165,6 +165,16 @@ public abstract class StateMachineObjectSupport<S, E> extends LifecycleObjectSup
}
}
protected void stateChangedInRelay() {
// TODO: this is a temporary tweak to know when state is
// changed in a submachine/regions order to give
// state machine a change to request executor login again
// which is needed when we use multiple thread. with multiple
// threads submachines may do their stuff after thread handling
// main machine has already finished its execution logic, thus
// re-scheduling is needed.
}
/**
* This class is used to relay listener events from a submachines which works
* as its own listener context. User only connects to main root machine and
@@ -175,6 +185,7 @@ public abstract class StateMachineObjectSupport<S, E> extends LifecycleObjectSup
@Override
public void stateChanged(State<S, E> from, State<S, E> to) {
stateListener.stateChanged(from, to);
stateChangedInRelay();
}
@Override

36
spring-statemachine-core/src/main/java/org/springframework/statemachine/transition/InitialTransition.java
View File

@@ -23,19 +23,51 @@ import org.springframework.statemachine.action.Action;
import org.springframework.statemachine.state.State;
import org.springframework.statemachine.trigger.Trigger;
/**
* {@link Transition} used during a state machine start.
*
* @author Janne Valkealahti
*
* @param <S> the type of state
* @param <E> the type of event
*/
public class InitialTransition<S, E> implements Transition<S, E> {
private final State<S, E> target;
private final Collection<Action<S, E>> actions;
/**
* Instantiates a new initial transition.
*
* @param target the initial target state
*/
public InitialTransition(State<S, E> target) {
this.target = target;
this.actions = null;
}
/**
* Instantiates a new initial transition.
*
* @param target the initial target state
* @param action the initial action
*/
public InitialTransition(State<S, E> target, Action<S, E> action) {
this.target = target;
ArrayList<Action<S,E>> list = new ArrayList<Action<S, E>>();
list.add(action);
if (action != null) {
list.add(action);
}
this.actions = list;
}
/**
* Instantiates a new initial transition.
*
* @param target the initial target state
* @param action the initial actions
*/
public InitialTransition(State<S, E> target, Collection<Action<S, E>> actions) {
this.target = target;
this.actions = actions;
@@ -73,7 +105,7 @@ public class InitialTransition<S, E> implements Transition<S, E> {
@Override
public TransitionKind getKind() {
return null;
return TransitionKind.INITIAL;
}
}

7
spring-statemachine-core/src/main/java/org/springframework/statemachine/transition/TransitionKind.java
View File

@@ -18,7 +18,7 @@ package org.springframework.statemachine.transition;
/**
* Defines enumeration of a {@link Transition} kind. This is uses within a
* transition to indicate whether its type is external, internal or local.
*
*
* @author Janne Valkealahti
*
*/
@@ -31,6 +31,9 @@ public enum TransitionKind {
INTERNAL,
/** Indicates a local transition kind. */
LOCAL
LOCAL,
/** Indicates an initial transition kind. */
INITIAL
}

67
spring-statemachine-core/src/test/java/org/springframework/statemachine/AbstractStateMachineTests.java
View File

@@ -30,9 +30,11 @@ import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;
import org.springframework.statemachine.StateContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.statemachine.action.Action;
import org.springframework.statemachine.guard.Guard;
import org.springframework.statemachine.listener.StateMachineListenerAdapter;
import org.springframework.statemachine.state.State;
/**
* Base class for stace machine tests.
@@ -104,6 +106,23 @@ public abstract class AbstractStateMachineTests {
}
@Configuration
public static class BaseConfig2 {
@Bean
public TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(5);
return taskExecutor;
}
@Bean
public TaskScheduler taskScheduler() {
return new ConcurrentTaskScheduler();
}
}
public static class TestEntryAction extends AbstractTestAction {
public TestEntryAction() {
@@ -141,6 +160,30 @@ public abstract class AbstractStateMachineTests {
public static class TestAction extends AbstractTestAction {
}
public static class TestSleepAction extends AbstractTestAction {
long sleep;
long now;
public TestSleepAction(long sleep) {
super();
this.sleep = sleep;
}
@Override
public void execute(StateContext<TestStates, TestEvents> context) {
now = System.currentTimeMillis();
if (sleep > 0) {
try {
Thread.sleep(sleep);
} catch (InterruptedException e) {
}
}
super.execute(context);
}
}
public static class TestGuard implements Guard<TestStates, TestEvents> {
public CountDownLatch onEvaluateLatch = new CountDownLatch(1);
@@ -183,4 +226,26 @@ public abstract class AbstractStateMachineTests {
}
}
protected static class TestStateMachineListener extends StateMachineListenerAdapter<TestStates, TestEvents> {
volatile CountDownLatch stateChangedLatch = new CountDownLatch(0);
volatile CountDownLatch stateMachineStartedLatch = new CountDownLatch(3);
@Override
public void stateChanged(State<TestStates, TestEvents> from, State<TestStates, TestEvents> to) {
stateChangedLatch.countDown();
}
@Override
public void stateMachineStarted(StateMachine<TestStates, TestEvents> stateMachine) {
stateMachineStartedLatch.countDown();
}
void reset(int c1, int c2) {
stateChangedLatch = new CountDownLatch(c1);
stateMachineStartedLatch = new CountDownLatch(c2);
}
}
}

4
spring-statemachine-core/src/test/java/org/springframework/statemachine/EnumStateMachineTests.java
View File

@@ -173,7 +173,8 @@ public class EnumStateMachineTests extends AbstractStateMachineTests {
@Test
public void testInternalTransitions() {
State<TestStates,TestEvents> stateSI = new EnumState<TestStates,TestEvents>(TestStates.SI);
PseudoState<TestStates,TestEvents> pseudoState = new DefaultPseudoState<TestStates,TestEvents>(PseudoStateKind.INITIAL);
State<TestStates,TestEvents> stateSI = new EnumState<TestStates,TestEvents>(TestStates.SI, pseudoState);
Collection<State<TestStates,TestEvents>> states = new ArrayList<State<TestStates,TestEvents>>();
states.add(stateSI);
@@ -190,6 +191,7 @@ public class EnumStateMachineTests extends AbstractStateMachineTests {
SyncTaskExecutor taskExecutor = new SyncTaskExecutor();
EnumStateMachine<TestStates, TestEvents> machine = new EnumStateMachine<TestStates, TestEvents>(states, transitions, stateSI);
machine.setTaskExecutor(taskExecutor);
machine.afterPropertiesSet();
machine.start();
machine.sendEvent(MessageBuilder.withPayload(TestEvents.E1).build());

211
spring-statemachine-core/src/test/java/org/springframework/statemachine/RegionMachineTests.java
View File

@@ -17,14 +17,15 @@ package org.springframework.statemachine;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.lessThan;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
@@ -38,7 +39,6 @@ import org.springframework.statemachine.config.EnumStateMachineConfigurerAdapter
import org.springframework.statemachine.config.builders.StateMachineStateConfigurer;
import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer;
import org.springframework.statemachine.event.StateMachineEventPublisherConfiguration;
import org.springframework.statemachine.listener.StateMachineListenerAdapter;
import org.springframework.statemachine.region.Region;
import org.springframework.statemachine.state.DefaultPseudoState;
import org.springframework.statemachine.state.EnumState;
@@ -47,6 +47,7 @@ import org.springframework.statemachine.state.PseudoStateKind;
import org.springframework.statemachine.state.RegionState;
import org.springframework.statemachine.state.State;
import org.springframework.statemachine.transition.DefaultExternalTransition;
import org.springframework.statemachine.transition.InitialTransition;
import org.springframework.statemachine.transition.Transition;
import org.springframework.statemachine.trigger.EventTrigger;
@@ -101,7 +102,8 @@ public class RegionMachineTests extends AbstractStateMachineTests {
transitions.add(transitionFromS2ToS3);
SyncTaskExecutor taskExecutor = new SyncTaskExecutor();
EnumStateMachine<TestStates, TestEvents> machine = new EnumStateMachine<TestStates, TestEvents>(states, transitions, stateSI);
Transition<TestStates,TestEvents> initialTransition = new InitialTransition<TestStates,TestEvents>(stateSI);
EnumStateMachine<TestStates, TestEvents> machine = new EnumStateMachine<TestStates, TestEvents>(states, transitions, stateSI, initialTransition, null, null);
machine.setTaskExecutor(taskExecutor);
machine.afterPropertiesSet();
machine.start();
@@ -130,7 +132,7 @@ public class RegionMachineTests extends AbstractStateMachineTests {
public void testMultiRegionBuildRaw() throws Exception {
SyncTaskExecutor taskExecutor = new SyncTaskExecutor();
PseudoState<TestStates,TestEvents> pseudoState = new DefaultPseudoState<TestStates,TestEvents>(PseudoStateKind.INITIAL);
State<TestStates,TestEvents> stateSI = new EnumState<TestStates,TestEvents>(TestStates.SI);
State<TestStates,TestEvents> stateSI = new EnumState<TestStates,TestEvents>(TestStates.SI, pseudoState);
TestEntryAction entryActionS111 = new TestEntryAction("S111");
TestExitAction exitActionS111 = new TestExitAction("S111");
@@ -164,7 +166,8 @@ public class RegionMachineTests extends AbstractStateMachineTests {
DefaultExternalTransition<TestStates,TestEvents> transitionFromS111ToS112 =
new DefaultExternalTransition<TestStates,TestEvents>(stateS111, stateS112, null, TestEvents.E2, null, new EventTrigger<TestStates,TestEvents>(TestEvents.E2));
transitions11.add(transitionFromS111ToS112);
EnumStateMachine<TestStates, TestEvents> machine11 = new EnumStateMachine<TestStates, TestEvents>(states11, transitions11, stateS111);
Transition<TestStates,TestEvents> initialTransition11 = new InitialTransition<TestStates,TestEvents>(stateS111);
EnumStateMachine<TestStates, TestEvents> machine11 = new EnumStateMachine<TestStates, TestEvents>(states11, transitions11, stateS111, initialTransition11, null, null);
machine11.setTaskExecutor(taskExecutor);
machine11.afterPropertiesSet();
@@ -175,7 +178,8 @@ public class RegionMachineTests extends AbstractStateMachineTests {
DefaultExternalTransition<TestStates,TestEvents> transitionFromSIToS121 =
new DefaultExternalTransition<TestStates,TestEvents>(stateSI, stateS111, null, TestEvents.E3, null, new EventTrigger<TestStates,TestEvents>(TestEvents.E3));
transitions12.add(transitionFromSIToS121);
EnumStateMachine<TestStates, TestEvents> machine12 = new EnumStateMachine<TestStates, TestEvents>(states12, transitions12, stateS121);
Transition<TestStates,TestEvents> initialTransition12 = new InitialTransition<TestStates,TestEvents>(stateS121);
EnumStateMachine<TestStates, TestEvents> machine12 = new EnumStateMachine<TestStates, TestEvents>(states12, transitions12, stateS121, initialTransition12, null, null);
machine12.setTaskExecutor(taskExecutor);
machine12.afterPropertiesSet();
@@ -190,7 +194,8 @@ public class RegionMachineTests extends AbstractStateMachineTests {
DefaultExternalTransition<TestStates,TestEvents> transitionFromSIToRegionstate =
new DefaultExternalTransition<TestStates,TestEvents>(stateSI, stateR, null, TestEvents.E1, null, new EventTrigger<TestStates,TestEvents>(TestEvents.E1));
transitions.add(transitionFromSIToRegionstate);
EnumStateMachine<TestStates, TestEvents> machine = new EnumStateMachine<TestStates, TestEvents>(states, transitions, stateR);
Transition<TestStates,TestEvents> initialTransition = new InitialTransition<TestStates,TestEvents>(stateR);
EnumStateMachine<TestStates, TestEvents> machine = new EnumStateMachine<TestStates, TestEvents>(states, transitions, stateR, initialTransition, null, null);
machine.setTaskExecutor(taskExecutor);
machine.afterPropertiesSet();
@@ -221,7 +226,7 @@ public class RegionMachineTests extends AbstractStateMachineTests {
@Test
public void testMultiRegion() throws Exception {
context.register(BaseConfig.class, StateMachineEventPublisherConfiguration.class, Config1.class);
context.register(StateMachineEventPublisherConfiguration.class, Config1.class);
context.refresh();
assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE));
@SuppressWarnings("unchecked")
@@ -253,7 +258,7 @@ public class RegionMachineTests extends AbstractStateMachineTests {
@SuppressWarnings("unchecked")
@Test
public void testRegionsInNestedState() throws Exception {
context.register(BaseConfig.class, StateMachineEventPublisherConfiguration.class, Config2.class);
context.register(StateMachineEventPublisherConfiguration.class, Config2.class);
context.refresh();
EnumStateMachine<TestStates,TestEvents> machine =
context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, EnumStateMachine.class);
@@ -261,11 +266,85 @@ public class RegionMachineTests extends AbstractStateMachineTests {
Collection<Object> states = TestUtils.readField("states", machine);
assertThat(states.size(), is(3));
assertThat(states, containsInAnyOrder(instanceOf(EnumState.class), instanceOf(EnumState.class), instanceOf(RegionState.class)));
TestStateMachineListener listener = context.getBean(TestStateMachineListener.class);
machine.addStateListener(listener);
machine.start();
listener.reset(3, 0);
machine.sendEvent(TestEvents.E1);
assertThat(listener.stateChangedLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S2, TestStates.S20, TestStates.S30));
}
@Test
public void testParallelRegionExecution() throws Exception {
context.register(StateMachineEventPublisherConfiguration.class, Config3.class, BaseConfig2.class);
context.refresh();
assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE));
@SuppressWarnings("unchecked")
EnumStateMachine<TestStates,TestEvents> machine =
context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, EnumStateMachine.class);
assertThat(machine, notNullValue());
TestSleepAction action1 = context.getBean("action1", TestSleepAction.class);
TestSleepAction action2 = context.getBean("action2", TestSleepAction.class);
TestStateMachineListener listener = context.getBean(TestStateMachineListener.class);
machine.addStateListener(listener);
machine.start();
assertThat(listener.stateMachineStartedLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S10, TestStates.S20));
listener.reset(2, 0);
machine.sendEvent(TestEvents.E1);
assertThat(listener.stateChangedLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S11, TestStates.S21));
listener.reset(1, 0);
machine.sendEvent(TestEvents.E2);
assertThat(listener.stateChangedLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S10, TestStates.S21));
listener.reset(1, 0);
machine.sendEvent(TestEvents.E3);
assertThat(listener.stateChangedLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S10, TestStates.S20));
// check that actions are called and that both are executed
// within a time which is less than their sleep time,
// indicating that we must have paralled execution
assertThat(action1.now, greaterThan(0l));
assertThat(action2.now, greaterThan(0l));
assertThat(Math.abs(action1.now-action2.now), lessThan(1999l));
}
@Test
public void testParallelRegionExecutionInInitialState() throws Exception {
context.register(StateMachineEventPublisherConfiguration.class, Config4.class, BaseConfig2.class);
context.refresh();
assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE));
@SuppressWarnings("unchecked")
EnumStateMachine<TestStates,TestEvents> machine =
context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, EnumStateMachine.class);
assertThat(machine, notNullValue());
TestSleepAction action1 = context.getBean("action1", TestSleepAction.class);
TestSleepAction action2 = context.getBean("action2", TestSleepAction.class);
TestStateMachineListener listener = context.getBean(TestStateMachineListener.class);
machine.addStateListener(listener);
machine.start();
assertThat(listener.stateMachineStartedLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S10, TestStates.S20));
listener.reset(2, 0);
machine.sendEvent(TestEvents.E1);
assertThat(listener.stateChangedLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S11, TestStates.S21));
// check that actions are called and that both are executed
// within a time which is less than their sleep time,
// indicating that we must have paralled execution
assertThat(action1.now, greaterThan(0l));
assertThat(action2.now, greaterThan(0l));
assertThat(Math.abs(action1.now-action2.now), lessThan(1999l));
}
@Configuration
@EnableStateMachine
static class Config1 extends EnumStateMachineConfigurerAdapter<TestStates, TestEvents> {
@@ -350,27 +429,117 @@ public class RegionMachineTests extends AbstractStateMachineTests {
.event(TestEvents.E1);
}
@Bean
public TestStateMachineListener testStateMachineListener() {
return new TestStateMachineListener();
}
}
private static class TestStateMachineListener extends StateMachineListenerAdapter<TestStates, TestEvents> {
volatile CountDownLatch stateChangedLatch = new CountDownLatch(0);
volatile CountDownLatch stateMachineStartedLatch = new CountDownLatch(3);
@Configuration
@EnableStateMachine
static class Config3 extends EnumStateMachineConfigurerAdapter<TestStates, TestEvents> {
@Override
public void stateChanged(State<TestStates, TestEvents> from, State<TestStates, TestEvents> to) {
stateChangedLatch.countDown();
public void configure(StateMachineStateConfigurer<TestStates, TestEvents> states) throws Exception {
states
.withStates()
.initial(TestStates.S10)
.state(TestStates.S10)
.state(TestStates.S11, action1(), null)
.and()
.withStates()
.initial(TestStates.S20)
.state(TestStates.S20)
.state(TestStates.S21, action2(), null);
}
@Override
public void stateMachineStarted(StateMachine<TestStates, TestEvents> stateMachine) {
stateMachineStartedLatch.countDown();
public void configure(StateMachineTransitionConfigurer<TestStates, TestEvents> transitions) throws Exception {
transitions
.withExternal()
.source(TestStates.S10)
.target(TestStates.S11)
.event(TestEvents.E1)
.and()
.withExternal()
.source(TestStates.S11)
.target(TestStates.S10)
.event(TestEvents.E2)
.and()
.withExternal()
.source(TestStates.S20)
.target(TestStates.S21)
.event(TestEvents.E1)
.and()
.withExternal()
.source(TestStates.S21)
.target(TestStates.S20)
.event(TestEvents.E3);
}
void reset(int c1, int c2) {
stateChangedLatch = new CountDownLatch(c1);
stateMachineStartedLatch = new CountDownLatch(c2);
@Bean
public TestStateMachineListener testStateMachineListener() {
return new TestStateMachineListener();
}
@Bean
public TestSleepAction action1() {
return new TestSleepAction(2000);
}
@Bean
public TestSleepAction action2() {
return new TestSleepAction(2000);
}
}
@Configuration
@EnableStateMachine
static class Config4 extends EnumStateMachineConfigurerAdapter<TestStates, TestEvents> {
@Override
public void configure(StateMachineStateConfigurer<TestStates, TestEvents> states) throws Exception {
states
.withStates()
.initial(TestStates.S10)
.state(TestStates.S10, action1(), null)
.state(TestStates.S11)
.and()
.withStates()
.initial(TestStates.S20)
.state(TestStates.S20, action2(), null)
.state(TestStates.S21);
}
@Override
public void configure(StateMachineTransitionConfigurer<TestStates, TestEvents> transitions) throws Exception {
transitions
.withExternal()
.source(TestStates.S10)
.target(TestStates.S11)
.event(TestEvents.E1)
.and()
.withExternal()
.source(TestStates.S20)
.target(TestStates.S21)
.event(TestEvents.E1);
}
@Bean
public TestStateMachineListener testStateMachineListener() {
return new TestStateMachineListener();
}
@Bean
public TestSleepAction action1() {
return new TestSleepAction(2000);
}
@Bean
public TestSleepAction action2() {
return new TestSleepAction(2000);
}
}

7
spring-statemachine-core/src/test/java/org/springframework/statemachine/RelayTests.java
View File

@@ -15,10 +15,13 @@
*/
package org.springframework.statemachine;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.Assert.assertThat;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
@@ -44,8 +47,12 @@ public class RelayTests extends AbstractStateMachineTests {
EnumStateMachine<TestStates,TestEvents> machine =
context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, EnumStateMachine.class);
assertThat(machine, notNullValue());
TestStateMachineListener listener = new TestStateMachineListener();
machine.addStateListener(listener);
machine.start();
listener.reset(3, 0);
machine.sendEvent(TestEvents.E1);
assertThat(listener.stateChangedLatch.await(5, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), contains(TestStates.S2, TestStates.S21));
}

11
spring-statemachine-core/src/test/java/org/springframework/statemachine/SubStateMachineTests.java
View File

@@ -131,9 +131,11 @@ public class SubStateMachineTests extends AbstractStateMachineTests {
SyncTaskExecutor taskExecutor = new SyncTaskExecutor();
machine.setTaskExecutor(taskExecutor);
machine.afterPropertiesSet();
machine.start();
submachine1.setTaskExecutor(taskExecutor);
submachine1.afterPropertiesSet();
submachine11.setTaskExecutor(taskExecutor);
submachine11.afterPropertiesSet();
machine.start();
machine.sendEvent(TestEvents.E1);
@@ -221,8 +223,9 @@ public class SubStateMachineTests extends AbstractStateMachineTests {
SyncTaskExecutor taskExecutor = new SyncTaskExecutor();
machine.setTaskExecutor(taskExecutor);
machine.afterPropertiesSet();
machine.start();
submachine11.setTaskExecutor(taskExecutor);
submachine11.afterPropertiesSet();
machine.start();
machine.sendEvent(TestEvents.E1);
@@ -318,9 +321,11 @@ public class SubStateMachineTests extends AbstractStateMachineTests {
SyncTaskExecutor taskExecutor = new SyncTaskExecutor();
machine.setTaskExecutor(taskExecutor);
machine.afterPropertiesSet();
machine.start();
submachine1.setTaskExecutor(taskExecutor);
submachine1.afterPropertiesSet();
submachine11.setTaskExecutor(taskExecutor);
submachine11.afterPropertiesSet();
machine.start();
machine.sendEvent(TestEvents.E1);

5
spring-statemachine-core/src/test/java/org/springframework/statemachine/state/RegionStateTests.java
View File

@@ -41,8 +41,8 @@ public class RegionStateTests extends AbstractStateMachineTests {
@Test
public void testSimpleRegionState() {
State<TestStates,TestEvents> stateSI = new EnumState<TestStates,TestEvents>(TestStates.SI);
PseudoState<TestStates,TestEvents> pseudoState = new DefaultPseudoState<TestStates,TestEvents>(PseudoStateKind.INITIAL);
State<TestStates,TestEvents> stateSI = new EnumState<TestStates,TestEvents>(TestStates.SI, pseudoState);
State<TestStates,TestEvents> stateS1 = new EnumState<TestStates,TestEvents>(TestStates.S1);
State<TestStates,TestEvents> stateS2 = new EnumState<TestStates,TestEvents>(TestStates.S2);
State<TestStates,TestEvents> stateS3 = new EnumState<TestStates,TestEvents>(TestStates.S3);
@@ -71,6 +71,7 @@ public class RegionStateTests extends AbstractStateMachineTests {
SyncTaskExecutor taskExecutor = new SyncTaskExecutor();
EnumStateMachine<TestStates, TestEvents> machine = new EnumStateMachine<TestStates, TestEvents>(states, transitions, stateSI);
machine.setTaskExecutor(taskExecutor);
machine.afterPropertiesSet();
machine.start();
Collection<Region<TestStates,TestEvents>> regions = new ArrayList<Region<TestStates,TestEvents>>();

5
spring-statemachine-core/src/test/java/org/springframework/statemachine/state/SubmachineStateTests.java
View File

@@ -55,8 +55,8 @@ public class SubmachineStateTests extends AbstractStateMachineTests {
@Test
public void testSimpleSubmachineState() {
State<TestStates,TestEvents> stateSI = new EnumState<TestStates,TestEvents>(TestStates.SI);
PseudoState<TestStates,TestEvents> pseudoState = new DefaultPseudoState<TestStates,TestEvents>(PseudoStateKind.INITIAL);
State<TestStates,TestEvents> stateSI = new EnumState<TestStates,TestEvents>(TestStates.SI, pseudoState);
State<TestStates,TestEvents> stateS1 = new EnumState<TestStates,TestEvents>(TestStates.S1);
State<TestStates,TestEvents> stateS2 = new EnumState<TestStates,TestEvents>(TestStates.S2);
State<TestStates,TestEvents> stateS3 = new EnumState<TestStates,TestEvents>(TestStates.S3);
@@ -85,6 +85,7 @@ public class SubmachineStateTests extends AbstractStateMachineTests {
SyncTaskExecutor taskExecutor = new SyncTaskExecutor();
EnumStateMachine<TestStates, TestEvents> machine = new EnumStateMachine<TestStates, TestEvents>(states, transitions, stateSI);
machine.setTaskExecutor(taskExecutor);
machine.afterPropertiesSet();
machine.start();
StateMachineState<TestStates,TestEvents> state = new StateMachineState<TestStates,TestEvents>(TestStates.S4, machine);

2
spring-statemachine-samples/showcase/src/test/java/demo/showcase/ShowcaseTests.java
View File

@@ -76,7 +76,7 @@ public class ShowcaseTests {
assertThat(listener.statesExited.size(), is(2));
assertThat(listener.statesExited.get(0).getId(), is(States.S11));
assertThat(listener.statesExited.get(1).getId(), is(States.S1));
assertThat(listener.transitionCount, is(1));
assertThat(listener.transitionCount, is(2));
}
@Test

12
spring-statemachine-samples/tasks/src/main/java/demo/tasks/Application.java
View File

@@ -8,6 +8,8 @@ import java.util.Map;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.shell.Bootstrap;
import org.springframework.statemachine.StateContext;
import org.springframework.statemachine.action.Action;
@@ -127,8 +129,7 @@ public class Application {
@Override
public void execute(StateContext<States, Events> context) {
Map<Object, Object> variables = context.getExtendedState().getVariables();
if (variables.get("T1").equals(false)) {
variables.put("T1", true);
if (variables.get("T1").equals(true)) {
context.getStateMachine().sendEvent(Events.CONTINUE);
} else {
context.getStateMachine().sendEvent(Events.FALLBACK);
@@ -157,6 +158,13 @@ public class Application {
return new Tasks();
}
@Bean
public TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(5);
return taskExecutor;
}
}
//end::snippetA[]

13
spring-statemachine-samples/tasks/src/main/java/demo/tasks/Tasks.java
View File

@@ -54,19 +54,25 @@ public class Tasks {
@StatesOnTransition(target = States.T1)
public void taskT1(ExtendedState extendedState) {
log.info("run task on T1");
sleep(2000);
extendedState.getVariables().put("T1", tasks.get("T1"));
log.info("run task on T1 done");
}
@StatesOnTransition(target = States.T2)
public void taskT2(ExtendedState extendedState) {
log.info("run task on T2");
sleep(2000);
extendedState.getVariables().put("T2", tasks.get("T2"));
log.info("run task on T2 done");
}
@StatesOnTransition(target = States.T3)
public void taskT3(ExtendedState extendedState) {
log.info("run task on T3");
sleep(2000);
extendedState.getVariables().put("T3", tasks.get("T3"));
log.info("run task on T3 done");
}
@StatesOnTransition(target = States.AUTOMATIC)
@@ -87,6 +93,13 @@ public class Tasks {
}
}
private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
}
}
@Override
public String toString() {
return "Tasks " + tasks;

18
spring-statemachine-samples/tasks/src/test/java/demo/tasks/TasksTests.java
View File

@@ -46,22 +46,22 @@ public class TasksTests {
@Test
public void testRunOnce() throws InterruptedException {
listener.reset(3, 0, 0);
listener.reset(9, 0, 0);
tasks.run();
assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
assertThat(listener.stateChangedLatch.await(6, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), contains(States.READY));
}
@Test
public void testRunTwice() throws InterruptedException {
listener.reset(3, 0, 0);
listener.reset(9, 0, 0);
tasks.run();
assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
assertThat(listener.stateChangedLatch.await(6, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), contains(States.READY));
listener.reset(3, 0, 0);
listener.reset(9, 0, 0);
tasks.run();
assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
assertThat(listener.stateChangedLatch.await(6, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), contains(States.READY));
}
@@ -70,19 +70,19 @@ public class TasksTests {
listener.reset(11, 0, 0);
tasks.fail("T1");
tasks.run();
assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
assertThat(listener.stateChangedLatch.await(6, TimeUnit.SECONDS), is(true));
assertThat(listener.stateChangedCount, is(11));
assertThat(machine.getState().getIds(), contains(States.READY));
}
@Test
public void testFailManualFix() throws InterruptedException {
listener.reset(3, 0, 0);
listener.reset(9, 0, 0);
tasks.fail("T2");
tasks.run();
tasks.fix("T2");
tasks.cont();
assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
assertThat(listener.stateChangedLatch.await(6, TimeUnit.SECONDS), is(true));
assertThat(machine.getState().getIds(), contains(States.READY));
}