Support executing state action with timeout

- Add new concepts around executing state do actions
  allowing to use timeouts via StateDoActionPolicy.
- Global config per machine or via event headers.
- Backport #501
- Relates #504

docs/src/reference/asciidoc/sm.adoc

@@ -232,8 +232,36 @@ has been completed.
 State Actions are executed using a normal Spring `TaskScheduler`
 wrapped within a `Runnable` which may get cancelled via
 `ScheduledFuture`. What this means is that whatever you're doing in your
-action, you need to be able to catch `InterruptedException` which is
-raised if task is cancelled.
+action, you need to be able to catch `InterruptedException` or generally
+periodically check if _Thread_ is interrupted.
+
+Below shows typical config which uses default _IMMEDIATE_CANCEL_ which
+would simply cancel running task immediately when state is complete.
+
+[source,java,indent=0]
+----
+include::samples/DocsConfigurationSampleTests11.java[tags=snippetA]
+----
+
+Policy can be set to _TIMEOUT_CANCEL_ together with a global timeout
+per machine. This changes state behaviour to wait action completion
+before cancel is requested.
+
+[source,java,indent=0]
+----
+include::samples/DocsConfigurationSampleTests11.java[tags=snippetB]
+----
+
+If _Event_ directly take machine into a state so that event headers
+are available to particular action, it is also possible to use dedicated
+event header to instruct a specific timeout which is defined in _millis_.
+Reserved header value _StateMachineMessageHeaders.HEADER_DO_ACTION_TIMEOUT_
+is used for this.
+
+[source,java,indent=0]
+----
+include::samples/DocsConfigurationSampleTests11.java[tags=snippetC]
+----
 
 [[statemachine-config-transition-actions-errorhandling]]
 ==== Transition Action Error Handling

spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineMessageHeaders.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2018 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;
+
+import org.springframework.messaging.Message;
+
+/**
+ * Reserved {@link Message} headers used in {@link StateMachine} events.
+ *
+ * @author Janne Valkealahti
+ *
+ */
+public final class StateMachineMessageHeaders {
+
+	/**
+	 * Header value defining state do action timeout. Format is a number
+	 * defining timeout as {@code millis}.
+	 */
+	public final static String HEADER_DO_ACTION_TIMEOUT = "STATEMACHINE_DO_ACTION_TIMEOUT";
+}

spring-statemachine-core/src/main/java/org/springframework/statemachine/action/StateDoActionPolicy.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2018 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.action;
+
+/**
+ * Enumerations for possible state do action policies. {@code IMMEDIATE_CANCEL}
+ * is a default setting.
+ *
+ * @author Janne Valkealahti
+ *
+ */
+public enum StateDoActionPolicy {
+
+	/**
+	 * Policy interrupting action immediately when state is exited.
+	 */
+	IMMEDIATE_CANCEL,
+
+	/**
+	 * Policy interrupting action after a timeout before state is exited.
+	 */
+	TIMEOUT_CANCEL;
+}

spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java

@@ -232,8 +232,10 @@ public abstract class AbstractStateMachineFactory<S, E> extends LifecycleObjectS
 					regions.add(pop.machine);
 				}
 				S parent = (S)peek.getParent();
-				RegionState<S, E> rstate = buildRegionStateInternal(parent, regions, null, stateData != null ? stateData.getEntryActions() : null,
-						stateData != null ? stateData.getExitActions() : null, new DefaultPseudoState<S, E>(PseudoStateKind.INITIAL));
+				RegionState<S, E> rstate = buildRegionStateInternal(parent, regions, null,
+						stateData != null ? stateData.getEntryActions() : null,
+						stateData != null ? stateData.getExitActions() : null,
+						new DefaultPseudoState<S, E>(PseudoStateKind.INITIAL), stateMachineModel);
 				if (stateData != null) {
 					stateMap.put(stateData.getState(), rstate);
 				} else {
@@ -580,8 +582,15 @@ public abstract class AbstractStateMachineFactory<S, E> extends LifecycleObjectS
 				if (stateData.isInitial()) {
 					pseudoState = new DefaultPseudoState<S, E>(PseudoStateKind.INITIAL);
 				}
-				state = new StateMachineState<S, E>(stateData.getState(), stateMachine, stateData.getDeferred(),
-						stateData.getEntryActions(), stateData.getExitActions(), pseudoState);
+				StateMachineState<S, E> stateMachineState = new StateMachineState<S, E>(stateData.getState(),
+						stateMachine, stateData.getDeferred(), stateData.getEntryActions(), stateData.getExitActions(),
+						pseudoState);
+				stateMachineState
+						.setStateDoActionPolicy(stateMachineModel.getConfigurationData().getStateDoActionPolicy());
+				stateMachineState.setStateDoActionPolicyTimeout(
+						stateMachineModel.getConfigurationData().getStateDoActionPolicyTimeout());
+				state = stateMachineState;
+
 				// TODO: below if/else doesn't feel right
 				if (stateDatas.size() > 1 && stateData.isInitial()) {
 					initialState = state;
@@ -901,7 +910,7 @@ public abstract class AbstractStateMachineFactory<S, E> extends LifecycleObjectS
 
 	protected abstract RegionState<S, E> buildRegionStateInternal(S id, Collection<Region<S, E>> regions, Collection<E> deferred,
 			Collection<? extends Action<S, E>> entryActions, Collection<? extends Action<S, E>> exitActions,
-			PseudoState<S, E> pseudoState);
+			PseudoState<S, E> pseudoState, StateMachineModel<S, E> stateMachineModel);
 
 	/**
 	 * Simple utility listener waiting machine to get started if

spring-statemachine-core/src/main/java/org/springframework/statemachine/config/ObjectStateMachineFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -112,13 +112,19 @@ public class ObjectStateMachineFactory<S, E> extends AbstractStateMachineFactory
 		if (taskScheduler != null) {
 			objectState.setTaskScheduler(taskScheduler);
 		}
+		objectState.setStateDoActionPolicy(stateMachineModel.getConfigurationData().getStateDoActionPolicy());
+		objectState.setStateDoActionPolicyTimeout(stateMachineModel.getConfigurationData().getStateDoActionPolicyTimeout());
 		return objectState;
 	}
 
 	@Override
 	protected RegionState<S, E> buildRegionStateInternal(S id, Collection<Region<S, E>> regions, Collection<E> deferred,
-			Collection<? extends Action<S, E>> entryActions, Collection<? extends Action<S, E>> exitActions, PseudoState<S, E> pseudoState) {
-		return new RegionState<S, E>(id, regions, deferred, entryActions, exitActions, pseudoState);
+			Collection<? extends Action<S, E>> entryActions, Collection<? extends Action<S, E>> exitActions,
+			PseudoState<S, E> pseudoState, StateMachineModel<S, E> stateMachineModel) {
+		RegionState<S,E> regionState = new RegionState<S, E>(id, regions, deferred, entryActions, exitActions, pseudoState);
+		regionState.setStateDoActionPolicy(stateMachineModel.getConfigurationData().getStateDoActionPolicy());
+		regionState.setStateDoActionPolicyTimeout(stateMachineModel.getConfigurationData().getStateDoActionPolicyTimeout());
+		return regionState;
 	}
 
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/config/builders/StateMachineConfigurationBuilder.java

@@ -22,6 +22,7 @@ import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.security.access.AccessDecisionManager;
+import org.springframework.statemachine.action.StateDoActionPolicy;
 import org.springframework.statemachine.config.common.annotation.AbstractConfiguredAnnotationBuilder;
 import org.springframework.statemachine.config.common.annotation.AnnotationBuilder;
 import org.springframework.statemachine.config.common.annotation.ObjectPostProcessor;
@@ -66,6 +67,8 @@ public class StateMachineConfigurationBuilder<S, E>
 	private TaskScheduler taskScheculer;
 	private boolean autoStart = false;
 	private TransitionConflictPolicy transitionConflictPolicy;
+	private StateDoActionPolicy stateDoActionPolicy;
+	private Long stateDoActionPolicyTimeout;
 	private StateMachineEnsemble<S, E> ensemble;
 	private final List<StateMachineListener<S, E>> listeners = new ArrayList<StateMachineListener<S, E>>();
 	private boolean securityEnabled = false;
@@ -146,8 +149,9 @@ public class StateMachineConfigurationBuilder<S, E>
 			}
 		}
 		return new ConfigurationData<S, E>(beanFactory, taskExecutor, taskScheculer, autoStart, ensemble, listeners,
-				securityEnabled, transitionSecurityAccessDecisionManager, eventSecurityAccessDecisionManager, eventSecurityRule,
-				transitionSecurityRule, verifierEnabled, verifier, machineId, stateMachineMonitor, interceptorsCopy, transitionConflictPolicy);
+				securityEnabled, transitionSecurityAccessDecisionManager, eventSecurityAccessDecisionManager,
+				eventSecurityRule, transitionSecurityRule, verifierEnabled, verifier, machineId, stateMachineMonitor,
+				interceptorsCopy, transitionConflictPolicy, stateDoActionPolicy, stateDoActionPolicyTimeout);
 	}
 
 	/**
@@ -303,4 +307,15 @@ public class StateMachineConfigurationBuilder<S, E>
 	public void setTransitionConflictPolicy(TransitionConflictPolicy transitionConflictPolicy) {
 		this.transitionConflictPolicy = transitionConflictPolicy;
 	}
+
+	/**
+	 * Sets the state do action policy and timeout.
+	 *
+	 * @param stateDoActionPolicy the new state do action policy
+	 * @param stateDoActionPolicyTimeout the new state do action policy timeout
+	 */
+	public void setStateDoActionPolicy(StateDoActionPolicy stateDoActionPolicy, Long stateDoActionPolicyTimeout) {
+		this.stateDoActionPolicy = stateDoActionPolicy;
+		this.stateDoActionPolicyTimeout = stateDoActionPolicyTimeout;
+	}
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/ConfigurationConfigurer.java

@@ -15,10 +15,13 @@
  */
 package org.springframework.statemachine.config.configurers;
 
+import java.util.concurrent.TimeUnit;
+
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.statemachine.StateMachine;
+import org.springframework.statemachine.action.StateDoActionPolicy;
 import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer;
 import org.springframework.statemachine.config.common.annotation.AnnotationConfigurerBuilder;
 import org.springframework.statemachine.listener.StateMachineListener;
@@ -88,10 +91,27 @@ public interface ConfigurationConfigurer<S, E> extends
 	ConfigurationConfigurer<S, E> listener(StateMachineListener<S, E> listener);
 
 	/**
-	 * Speficy a {@link TransitionConflictPolicy}. Default to {@link TransitionConflictPolicy#CHILD}.
+	 * Specify a {@link TransitionConflictPolicy}. Default to {@link TransitionConflictPolicy#CHILD}.
 	 *
 	 * @param transitionConflictPolicy the transition conflict policy
 	 * @return the configuration configurer
 	 */
 	ConfigurationConfigurer<S, E> transitionConflictPolicy(TransitionConflictPolicy transitionConflictPolicy);
+
+	/**
+	 * Specify a {@link StateDoActionPolicy}. Defaults to {@link StateDoActionPolicy#IMMEDIATE_CANCEL}.
+	 *
+	 * @param stateDoActionPolicy the state do action policy
+	 * @return the configuration configurer
+	 */
+	ConfigurationConfigurer<S, E> stateDoActionPolicy(StateDoActionPolicy stateDoActionPolicy);
+
+	/**
+	 * Specify a timeout used with {@link StateDoActionPolicy}.
+	 *
+	 * @param timeout the timeout
+	 * @param unit the time unit
+	 * @return the configuration configurer
+	 */
+	ConfigurationConfigurer<S, E> stateDoActionPolicyTimeout(long timeout, TimeUnit unit);
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/DefaultConfigurationConfigurer.java

@@ -17,10 +17,12 @@ package org.springframework.statemachine.config.configurers;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.scheduling.TaskScheduler;
+import org.springframework.statemachine.action.StateDoActionPolicy;
 import org.springframework.statemachine.config.builders.StateMachineConfigurationBuilder;
 import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer;
 import org.springframework.statemachine.config.common.annotation.AnnotationConfigurerAdapter;
@@ -46,6 +48,8 @@ public class DefaultConfigurationConfigurer<S, E>
 	private TaskScheduler taskScheculer;
 	private boolean autoStart = false;
 	private TransitionConflictPolicy transitionConflightPolicy;
+	private StateDoActionPolicy stateDoActionPolicy;
+	private Long stateDoActionPolicyTimeout;
 	private final List<StateMachineListener<S, E>> listeners = new ArrayList<StateMachineListener<S, E>>();
 
 	@Override
@@ -57,6 +61,7 @@ public class DefaultConfigurationConfigurer<S, E>
 		builder.setAutoStart(autoStart);
 		builder.setStateMachineListeners(listeners);
 		builder.setTransitionConflictPolicy(transitionConflightPolicy);
+		builder.setStateDoActionPolicy(stateDoActionPolicy, stateDoActionPolicyTimeout);
 	}
 
 	@Override
@@ -100,4 +105,16 @@ public class DefaultConfigurationConfigurer<S, E>
 		this.transitionConflightPolicy = transitionConflightPolicy;
 		return this;
 	}
+
+	@Override
+	public ConfigurationConfigurer<S, E> stateDoActionPolicy(StateDoActionPolicy stateDoActionPolicy) {
+		this.stateDoActionPolicy = stateDoActionPolicy;
+		return this;
+	}
+
+	@Override
+	public ConfigurationConfigurer<S, E> stateDoActionPolicyTimeout(long timeout, TimeUnit unit) {
+		this.stateDoActionPolicyTimeout = unit.toMillis(timeout);
+		return this;
+	}
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/config/model/ConfigurationData.java

@@ -24,6 +24,7 @@ import org.springframework.core.task.TaskExecutor;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;
 import org.springframework.security.access.AccessDecisionManager;
+import org.springframework.statemachine.action.StateDoActionPolicy;
 import org.springframework.statemachine.config.builders.StateMachineConfigurationBuilder;
 import org.springframework.statemachine.config.model.verifier.DefaultStateMachineModelVerifier;
 import org.springframework.statemachine.config.model.verifier.StateMachineModelVerifier;
@@ -50,6 +51,8 @@ public class ConfigurationData<S, E> {
 	private final TaskScheduler taskScheduler;
 	private final boolean autoStart;
 	private final TransitionConflictPolicy transitionConflictPolicy;
+	private final StateDoActionPolicy stateDoActionPolicy;
+	private final Long stateDoActionPolicyTimeout;
 	private final StateMachineEnsemble<S, E> ensemble;
 	private final List<StateMachineListener<S, E>> listeners;
 	private final boolean securityEnabled;
@@ -99,7 +102,7 @@ public class ConfigurationData<S, E> {
 			List<StateMachineInterceptor<S, E>> interceptors) {
 		this(beanFactory, taskExecutor, taskScheduler, autoStart, ensemble, listeners, securityEnabled,
 				transitionSecurityAccessDecisionManager, eventSecurityAccessDecisionManager, eventSecurityRule, transitionSecurityRule,
-				verifierEnabled, verifier, machineId, stateMachineMonitor, interceptors, null);
+				verifierEnabled, verifier, machineId, stateMachineMonitor, interceptors, null, null, null);
 	}
 
 	/**
@@ -122,6 +125,8 @@ public class ConfigurationData<S, E> {
 	 * @param stateMachineMonitor the state machine monitor
 	 * @param interceptors the state machine interceptors.
 	 * @param transitionConflightPolicy the transition conflict policy
+	 * @param stateDoActionPolicy the state do action policy
+	 * @param stateDoActionPolicyTimeout the state do action policy timeout
 	 */
 	public ConfigurationData(BeanFactory beanFactory, TaskExecutor taskExecutor,
 			TaskScheduler taskScheduler, boolean autoStart, StateMachineEnsemble<S, E> ensemble,
@@ -129,7 +134,8 @@ public class ConfigurationData<S, E> {
 			AccessDecisionManager transitionSecurityAccessDecisionManager, AccessDecisionManager eventSecurityAccessDecisionManager,
 			SecurityRule eventSecurityRule, SecurityRule transitionSecurityRule, boolean verifierEnabled,
 			StateMachineModelVerifier<S, E> verifier, String machineId, StateMachineMonitor<S, E> stateMachineMonitor,
-			List<StateMachineInterceptor<S, E>> interceptors, TransitionConflictPolicy transitionConflightPolicy) {
+			List<StateMachineInterceptor<S, E>> interceptors, TransitionConflictPolicy transitionConflightPolicy,
+			StateDoActionPolicy stateDoActionPolicy, Long stateDoActionPolicyTimeout) {
 		this.beanFactory = beanFactory;
 		this.taskExecutor = taskExecutor;
 		this.taskScheduler = taskScheduler;
@@ -147,6 +153,8 @@ public class ConfigurationData<S, E> {
 		this.stateMachineMonitor = stateMachineMonitor;
 		this.interceptors = interceptors;
 		this.transitionConflictPolicy = transitionConflightPolicy;
+		this.stateDoActionPolicy = stateDoActionPolicy;
+		this.stateDoActionPolicyTimeout = stateDoActionPolicyTimeout;
 	}
 
 	public String getMachineId() {
@@ -296,4 +304,22 @@ public class ConfigurationData<S, E> {
 	public TransitionConflictPolicy getTransitionConflictPolicy() {
 		return transitionConflictPolicy;
 	}
+
+	/**
+	 * Gets the state do action policy.
+	 *
+	 * @return the state do action policy
+	 */
+	public StateDoActionPolicy getStateDoActionPolicy() {
+		return stateDoActionPolicy;
+	}
+
+	/**
+	 * Gets the state do action policy timeout.
+	 *
+	 * @return the state do action policy timeout
+	 */
+	public Long getStateDoActionPolicyTimeout() {
+		return stateDoActionPolicyTimeout;
+	}
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/state/AbstractState.java

@@ -22,6 +22,7 @@ import java.util.Date;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.commons.logging.Log;
@@ -34,10 +35,12 @@ import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.action.Action;
 import org.springframework.statemachine.action.ActionListener;
 import org.springframework.statemachine.action.CompositeActionListener;
+import org.springframework.statemachine.action.StateDoActionPolicy;
 import org.springframework.statemachine.listener.StateMachineListener;
 import org.springframework.statemachine.listener.StateMachineListenerAdapter;
 import org.springframework.statemachine.region.Region;
 import org.springframework.statemachine.support.LifecycleObjectSupport;
+import org.springframework.statemachine.support.StateMachineUtils;
 import org.springframework.statemachine.trigger.Trigger;
 
 /**
@@ -61,9 +64,11 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
 	private final StateMachine<S, E> submachine;
 	private List<Trigger<S, E>> triggers = new ArrayList<Trigger<S, E>>();
 	private final CompositeStateListener<S, E> stateListener = new CompositeStateListener<S, E>();
-	private final List<ScheduledFuture<?>> cancellableActions = new ArrayList<>();
+	private final List<ScheduledAction> scheduledActions = new ArrayList<>();
 	private CompositeActionListener<S, E> actionListener;
 	private final List<StateMachineListener<S, E>> completionListeners = new CopyOnWriteArrayList<>();
+	private StateDoActionPolicy stateDoActionPolicy;
+	private Long stateDoActionPolicyTimeout;
 
 	/**
 	 * Instantiates a new abstract state.
@@ -399,6 +404,14 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
 		return triggers;
 	}
 
+	public void setStateDoActionPolicy(StateDoActionPolicy stateDoActionPolicy) {
+		this.stateDoActionPolicy = stateDoActionPolicy;
+	}
+
+	public void setStateDoActionPolicyTimeout(Long stateDoActionPolicyTimeout) {
+		this.stateDoActionPolicyTimeout = stateDoActionPolicyTimeout;
+	}
+
 	/**
 	 * Arm triggers.
 	 */
@@ -421,10 +434,30 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
 	 * Cancel existing state actions and clear list.
 	 */
 	protected void cancelStateActions() {
-		for (ScheduledFuture<?> future : cancellableActions) {
-			future.cancel(true);
+		if (log.isDebugEnabled()) {
+			log.debug("Handling finish of state actions, scheduledActions size is " + scheduledActions.size());
 		}
-		cancellableActions.clear();
+		for (ScheduledAction task : scheduledActions) {
+			if (task.timeout != null) {
+				if (log.isDebugEnabled()) {
+					log.debug("Timeouting scheduled state do action " + task);
+				}
+				try {
+					task.future.get(task.timeout, TimeUnit.MILLISECONDS);
+				} catch (Exception e) {
+					if (log.isDebugEnabled()) {
+						log.debug("Cancelling scheduled state do action after timeout " + task);
+					}
+					task.future.cancel(true);
+				}
+			} else {
+				if (log.isDebugEnabled()) {
+					log.debug("Cancelling scheduled state do action immediately " + task);
+				}
+				task.future.cancel(true);
+			}
+		}
+		scheduledActions.clear();
 	}
 
 	/**
@@ -440,8 +473,11 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
 		}
 		for (Action<S, E> action : stateActions) {
 			ScheduledFuture<?> future = scheduleAction(action, context, completionCount);
+			if (log.isDebugEnabled()) {
+				log.debug("Scheduling state do action " + action + " with future " + future);
+			}
 			if (future != null) {
-				cancellableActions.add(future);
+				scheduledActions.add(new ScheduledAction(future, resolveDoActionTimeout(context)));
 			}
 		}
 		if (isSimple() && stateActions.size() == 0) {
@@ -475,7 +511,8 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
 	 * @param completionCount the completion count tracker
 	 * @return the scheduled future
 	 */
-	protected ScheduledFuture<?> scheduleAction(final Action<S, E> action, final StateContext<S, E> context, final AtomicInteger completionCount) {
+	protected ScheduledFuture<?> scheduleAction(final Action<S, E> action, final StateContext<S, E> context,
+			final AtomicInteger completionCount) {
 		TaskScheduler taskScheduler = getTaskScheduler();
 		if (taskScheduler == null) {
 			log.error("Unable to schedule action as taskSchedule is not set, action=[" + action + "]");
@@ -498,6 +535,31 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
 		stateListener.onComplete(context);
 	}
 
+	private Long resolveDoActionTimeout(StateContext<S, E> context) {
+		Long timeout = null;
+		if (stateDoActionPolicy == StateDoActionPolicy.TIMEOUT_CANCEL) {
+			timeout = StateMachineUtils.getMessageHeaderDoActionTimeout(context);
+			if (timeout == null) {
+				timeout = stateDoActionPolicyTimeout;
+			}
+		}
+		return timeout;
+	}
+
+	private static class ScheduledAction {
+		ScheduledFuture<?> future;
+		Long timeout;
+
+		public ScheduledAction(ScheduledFuture<?> future, Long timeout) {
+			this.future = future;
+			this.timeout = timeout;
+		}
+		@Override
+		public String toString() {
+			return "ScheduledTask [future=" + future + ", timeout=" + timeout + "]";
+		}
+	}
+
 	@Override
 	public String toString() {
 		return "AbstractState [id=" + id + ", pseudoState=" + pseudoState + ", deferred=" + deferred + ", entryActions="

spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineUtils.java

@@ -18,6 +18,8 @@ package org.springframework.statemachine.support;
 import java.util.ArrayList;
 import java.util.Collection;
 
+import org.springframework.statemachine.StateContext;
+import org.springframework.statemachine.StateMachineMessageHeaders;
 import org.springframework.statemachine.state.PseudoState;
 import org.springframework.statemachine.state.PseudoStateKind;
 import org.springframework.statemachine.state.State;
@@ -170,4 +172,23 @@ public abstract class StateMachineUtils {
 		return false;
 	}
 
+	/**
+	 * Gets a particular message header from a {@link StateContext} and parses
+	 * it to {@code Long}. Any error parsing header results {@code NULL}.
+	 *
+	 * @param <S> the type of state
+	 * @param <E> the type of event
+	 * @param context the state context
+	 * @return the header as {@code Long} or {@code NULL}
+	 */
+	public static <S, E> Long getMessageHeaderDoActionTimeout(StateContext<S, E> context) {
+		try {
+			Number number = context.getMessageHeaders().get(StateMachineMessageHeaders.HEADER_DO_ACTION_TIMEOUT, Number.class);
+			if (number != null) {
+				return number.longValue();
+			}
+		} catch (Exception e) {
+		}
+		return null;
+	}
 }

spring-statemachine-core/src/test/java/org/springframework/statemachine/action/StateDoActivityActionTests.java

@@ -26,11 +26,14 @@ 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.messaging.support.MessageBuilder;
 import org.springframework.statemachine.AbstractStateMachineTests;
+import org.springframework.statemachine.StateMachineMessageHeaders;
 import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.StateMachineSystemConstants;
 import org.springframework.statemachine.config.EnableStateMachine;
 import org.springframework.statemachine.config.EnumStateMachineConfigurerAdapter;
+import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer;
 import org.springframework.statemachine.config.builders.StateMachineStateConfigurer;
 import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer;
 
@@ -247,6 +250,201 @@ public class StateDoActivityActionTests extends AbstractStateMachineTests {
 		}
 	}
 
+	@Test
+	public void testStateDoActionNotCancelledWithEventTimeout() throws Exception {
+		context.register(Config4.class);
+		context.refresh();
+		StateMachine<TestStates,TestEvents> machine =
+				context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class);
+		TestSleepAction testActionS2 = context.getBean("testActionS2", TestSleepAction.class);
+
+		assertThat(machine, notNullValue());
+		machine.start();
+
+		machine.sendEvent(MessageBuilder.withPayload(TestEvents.E1)
+				.setHeader(StateMachineMessageHeaders.HEADER_DO_ACTION_TIMEOUT, 4000).build());
+		assertThat(testActionS2.onExecuteStartLatch.await(2, TimeUnit.SECONDS), is(true));
+		machine.sendEvent(TestEvents.E2);
+		assertThat(testActionS2.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true));
+		assertThat(testActionS2.interruptedLatch.await(2, TimeUnit.SECONDS), is(false));
+	}
+
+	@Test
+	public void testStateDoActionCancelledWithEventTimeout() throws Exception {
+		context.register(Config4.class);
+		context.refresh();
+		StateMachine<TestStates,TestEvents> machine =
+				context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class);
+		TestSleepAction testActionS2 = context.getBean("testActionS2", TestSleepAction.class);
+
+		assertThat(machine, notNullValue());
+		machine.start();
+
+		machine.sendEvent(MessageBuilder.withPayload(TestEvents.E1)
+				.setHeader(StateMachineMessageHeaders.HEADER_DO_ACTION_TIMEOUT, 100).build());
+		assertThat(testActionS2.onExecuteStartLatch.await(2, TimeUnit.SECONDS), is(true));
+		machine.sendEvent(TestEvents.E2);
+		assertThat(testActionS2.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true));
+		assertThat(testActionS2.interruptedLatch.await(2, TimeUnit.SECONDS), is(true));
+	}
+
+	@Test
+	public void testStateDoActionCancelledWithConfigSetting() throws Exception {
+		context.register(Config5.class);
+		context.refresh();
+		StateMachine<TestStates,TestEvents> machine =
+				context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class);
+		TestSleepAction testActionS2 = context.getBean("testActionS2", TestSleepAction.class);
+
+		assertThat(machine, notNullValue());
+		machine.start();
+
+		machine.sendEvent(TestEvents.E1);
+		assertThat(testActionS2.onExecuteStartLatch.await(2, TimeUnit.SECONDS), is(true));
+		machine.sendEvent(TestEvents.E2);
+		assertThat(testActionS2.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true));
+		assertThat(testActionS2.interruptedLatch.await(2, TimeUnit.SECONDS), is(true));
+	}
+
+	@Test
+	public void testStateDoActionNotCancelledWithConfigTimeout() throws Exception {
+		context.register(Config6.class);
+		context.refresh();
+		StateMachine<TestStates,TestEvents> machine =
+				context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class);
+		TestSleepAction testActionS2 = context.getBean("testActionS2", TestSleepAction.class);
+
+		assertThat(machine, notNullValue());
+		machine.start();
+
+		machine.sendEvent(TestEvents.E1);
+		assertThat(testActionS2.onExecuteStartLatch.await(2, TimeUnit.SECONDS), is(true));
+		machine.sendEvent(TestEvents.E2);
+		assertThat(testActionS2.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true));
+		assertThat(testActionS2.interruptedLatch.await(2, TimeUnit.SECONDS), is(false));
+	}
+
+	@Configuration
+	@EnableStateMachine
+	static class Config4 extends EnumStateMachineConfigurerAdapter<TestStates, TestEvents> {
+
+		@Override
+		public void configure(StateMachineConfigurationConfigurer<TestStates, TestEvents> config) throws Exception {
+			config
+				.withConfiguration()
+					.stateDoActionPolicy(StateDoActionPolicy.TIMEOUT_CANCEL);
+		}
+
+		@Override
+		public void configure(StateMachineStateConfigurer<TestStates, TestEvents> states) throws Exception {
+			states
+				.withStates()
+					.initial(TestStates.S1)
+					.state(TestStates.S2, testActionS2())
+					.state(TestStates.S3);
+		}
+
+		@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.S2)
+					.target(TestStates.S3)
+					.event(TestEvents.E2);
+		}
+
+		@Bean
+		public TestSleepAction testActionS2() {
+			return new TestSleepAction(2000);
+		}
+	}
+
+	@Configuration
+	@EnableStateMachine
+	static class Config5 extends EnumStateMachineConfigurerAdapter<TestStates, TestEvents> {
+
+		@Override
+		public void configure(StateMachineConfigurationConfigurer<TestStates, TestEvents> config) throws Exception {
+			config
+				.withConfiguration()
+					.stateDoActionPolicy(StateDoActionPolicy.IMMEDIATE_CANCEL);
+		}
+
+		@Override
+		public void configure(StateMachineStateConfigurer<TestStates, TestEvents> states) throws Exception {
+			states
+				.withStates()
+					.initial(TestStates.S1)
+					.state(TestStates.S2, testActionS2())
+					.state(TestStates.S3);
+		}
+
+		@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.S2)
+					.target(TestStates.S3)
+					.event(TestEvents.E2);
+		}
+
+		@Bean
+		public TestSleepAction testActionS2() {
+			return new TestSleepAction(2000);
+		}
+	}
+
+	@Configuration
+	@EnableStateMachine
+	static class Config6 extends EnumStateMachineConfigurerAdapter<TestStates, TestEvents> {
+
+		@Override
+		public void configure(StateMachineConfigurationConfigurer<TestStates, TestEvents> config) throws Exception {
+			config
+				.withConfiguration()
+					.stateDoActionPolicy(StateDoActionPolicy.TIMEOUT_CANCEL)
+					.stateDoActionPolicyTimeout(10, TimeUnit.SECONDS);
+		}
+
+		@Override
+		public void configure(StateMachineStateConfigurer<TestStates, TestEvents> states) throws Exception {
+			states
+				.withStates()
+					.initial(TestStates.S1)
+					.state(TestStates.S2, testActionS2())
+					.state(TestStates.S3);
+		}
+
+		@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.S2)
+					.target(TestStates.S3)
+					.event(TestEvents.E2);
+		}
+
+		@Bean
+		public TestSleepAction testActionS2() {
+			return new TestSleepAction(2000);
+		}
+	}
+
 	@Override
 	protected AnnotationConfigApplicationContext buildContext() {
 		return new AnnotationConfigApplicationContext();

spring-statemachine-core/src/test/java/org/springframework/statemachine/docs/DocsConfigurationSampleTests11.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright 2018 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.docs;
+
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.statemachine.AbstractStateMachineTests;
+import org.springframework.statemachine.StateMachine;
+import org.springframework.statemachine.StateMachineMessageHeaders;
+import org.springframework.statemachine.action.StateDoActionPolicy;
+import org.springframework.statemachine.config.EnableStateMachine;
+import org.springframework.statemachine.config.StateMachineConfigurerAdapter;
+import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer;
+import org.springframework.statemachine.config.builders.StateMachineStateConfigurer;
+import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer;
+
+public class DocsConfigurationSampleTests11 extends AbstractStateMachineTests {
+
+// tag::snippetA[]
+	@Configuration
+	@EnableStateMachine
+	static class Config1 extends StateMachineConfigurerAdapter<String, String> {
+
+		@Override
+		public void configure(StateMachineConfigurationConfigurer<String, String> config) throws Exception {
+			config
+				.withConfiguration()
+					.stateDoActionPolicy(StateDoActionPolicy.IMMEDIATE_CANCEL);
+		}
+
+		@Override
+		public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
+			states
+				.withStates()
+					.initial("S1")
+					.state("S2", context -> {})
+					.state("S3");
+		}
+
+		@Override
+		public void configure(StateMachineTransitionConfigurer<String, String> transitions) throws Exception {
+			transitions
+				.withExternal()
+					.source("S1")
+					.target("S2")
+					.event("E1")
+					.and()
+				.withExternal()
+					.source("S2")
+					.target("S3")
+					.event("E2");
+		}
+	}
+// end::snippetA[]
+
+	@Configuration
+	@EnableStateMachine
+	static class Config2 extends StateMachineConfigurerAdapter<String, String> {
+
+// tag::snippetB[]
+		@Override
+		public void configure(StateMachineConfigurationConfigurer<String, String> config) throws Exception {
+			config
+				.withConfiguration()
+					.stateDoActionPolicy(StateDoActionPolicy.TIMEOUT_CANCEL)
+					.stateDoActionPolicyTimeout(10, TimeUnit.SECONDS);
+		}
+// end::snippetB[]
+
+		@Override
+		public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
+			states
+				.withStates()
+					.initial("S1")
+					.state("S2", context -> {})
+					.state("S3");
+		}
+
+		@Override
+		public void configure(StateMachineTransitionConfigurer<String, String> transitions) throws Exception {
+			transitions
+				.withExternal()
+					.source("S1")
+					.target("S2")
+					.event("E1")
+					.and()
+				.withExternal()
+					.source("S2")
+					.target("S3")
+					.event("E2");
+		}
+	}
+
+	static class Example1 {
+
+// tag::snippetC[]
+		@Autowired
+		StateMachine<String, String> stateMachine;
+
+		void sendEventUsingTimeout() {
+			stateMachine.sendEvent(MessageBuilder
+					.withPayload("E1")
+					.setHeader(StateMachineMessageHeaders.HEADER_DO_ACTION_TIMEOUT, 5000)
+					.build());
+
+		}
+// end::snippetC[]
+	}
+
+}