Enhance concepts around interceptor

- Rename StateChangeInterceptor to StateMachineInterceptor and
  add methods for pre/post transitions.
- Adding getId concept into a state machine to help distributed
  machines to know from where a state change originates.
- Fix ZookeeperStateMachineEnsemble to work better with
  persisted state machine contexts.

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

@@ -17,7 +17,7 @@ package org.springframework.statemachine;
 
 /**
  * Various constants used in state machine lib.
- * 
+ *
  * @author Janne Valkealahti
  *
  */
@@ -32,4 +32,7 @@ public abstract class StateMachineSystemConstants {
 	/** Default bean id for state machine event publisher. */
 	public static final String DEFAULT_ID_EVENT_PUBLISHER = "stateMachineEventPublisher";
 
+	/** State machine id key for headers and variables */
+	public static final String STATEMACHINE_IDENTIFIER = "_sm_id_";
+
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java

@@ -17,7 +17,7 @@ package org.springframework.statemachine.access;
 
 import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.StateMachineContext;
-import org.springframework.statemachine.support.StateChangeInterceptor;
+import org.springframework.statemachine.support.StateMachineInterceptor;
 
 /**
  * Functional interface exposing {@link StateMachine} internals.
@@ -44,11 +44,11 @@ public interface StateMachineAccess<S, E> {
 	void resetStateMachine(StateMachineContext<S, E> stateMachineContext);
 
 	/**
-	 * Adds the state change interceptor.
+	 * Adds the state machine interceptor.
 	 *
 	 * @param interceptor the interceptor
 	 */
-	void addStateChangeInterceptor(StateChangeInterceptor<S, E> interceptor);
+	void addStateMachineInterceptor(StateMachineInterceptor<S, E> interceptor);
 
 	/**
 	 * Sets if initial state is enabled when a state machine is

spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java

@@ -16,15 +16,16 @@
 package org.springframework.statemachine.ensemble;
 
 import java.util.Collection;
-import java.util.UUID;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.statemachine.ExtendedState;
+import org.springframework.statemachine.StateContext;
 import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.StateMachineContext;
+import org.springframework.statemachine.StateMachineSystemConstants;
 import org.springframework.statemachine.access.StateMachineAccess;
 import org.springframework.statemachine.access.StateMachineAccessor;
 import org.springframework.statemachine.access.StateMachineFunction;
@@ -32,7 +33,7 @@ import org.springframework.statemachine.listener.StateMachineListener;
 import org.springframework.statemachine.state.State;
 import org.springframework.statemachine.support.DefaultStateMachineContext;
 import org.springframework.statemachine.support.LifecycleObjectSupport;
-import org.springframework.statemachine.support.StateChangeInterceptor;
+import org.springframework.statemachine.support.StateMachineInterceptor;
 import org.springframework.statemachine.transition.Transition;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
@@ -53,11 +54,10 @@ import org.springframework.util.ObjectUtils;
 public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implements StateMachine<S, E> {
 
 	private final static Log log = LogFactory.getLog(DistributedStateMachine.class);
-	private final String uuid = UUID.randomUUID().toString();
 	private final StateMachineEnsemble<S, E> ensemble;
 	private final StateMachine<S, E> delegate;
 	private final LocalEnsembleListener listener = new LocalEnsembleListener();
-	private final LocalStateChangeInterceptor interceptor = new LocalStateChangeInterceptor();
+	private final LocalStateMachineInterceptor interceptor = new LocalStateMachineInterceptor();
 
 	/**
 	 * Instantiates a new distributed state machine.
@@ -78,7 +78,7 @@ public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implem
 
 			@Override
 			public void apply(StateMachineAccess<S, E> function) {
-				function.addStateChangeInterceptor(interceptor);
+				function.addStateMachineInterceptor(interceptor);
 			}
 		});
 	}
@@ -99,7 +99,10 @@ public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implem
 
 	@Override
 	public boolean sendEvent(Message<E> event) {
-		return delegate.sendEvent(MessageBuilder.fromMessage(event).setHeader("uuid", uuid).build());
+		// adding state machine id to the message so that
+		// listeners can know from where a state change originates
+		return delegate.sendEvent(MessageBuilder.fromMessage(event)
+				.setHeader(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER, delegate.getId()).build());
 	}
 
 	@Override
@@ -152,22 +155,49 @@ public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implem
 		return delegate.getStateMachineAccessor();
 	}
 
+	@Override
+	public String getId() {
+		return delegate.getId();
+	}
+
+	@Override
+	public String toString() {
+		return "DistributedStateMachine [delegate=" + delegate + "]";
+	}
+
 	/**
 	 * We intercept state changes order to attempt to update global
 	 * distributed state. This attempt is sent to an ensemble which will
 	 * tell us if that attempt was successful.
 	 */
-	private class LocalStateChangeInterceptor implements StateChangeInterceptor<S, E> {
+	private class LocalStateMachineInterceptor implements StateMachineInterceptor<S, E> {
 
 		@Override
 		public void preStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
 				StateMachine<S, E> stateMachine) {
-			if (message != null && ObjectUtils.nullSafeEquals(uuid, message.getHeaders().get("uuid"))) {
-				ensemble.setState(new DefaultStateMachineContext<S, E>(transition.getTarget()
-						.getId(), message.getPayload(), message.getHeaders(), stateMachine.getExtendedState()));
+			if (message != null
+					&& ObjectUtils.nullSafeEquals(delegate.getId(),
+							message.getHeaders().get(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER))) {
+				ensemble.setState(new DefaultStateMachineContext<S, E>(transition.getTarget().getId(), message
+						.getPayload(), message.getHeaders(), stateMachine.getExtendedState()));
 			}
 		}
 
+		@Override
+		public void postStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
+				StateMachine<S, E> stateMachine) {
+		}
+
+		@Override
+		public StateContext<S, E> preTransition(StateContext<S, E> stateContext) {
+			return stateContext;
+		}
+
+		@Override
+		public StateContext<S, E> postTransition(StateContext<S, E> stateContext) {
+			return stateContext;
+		}
+
 	}
 
 	/**

spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java

@@ -27,10 +27,25 @@ import org.springframework.statemachine.StateMachineContext;
  */
 public interface EnsembleListeger<S, E> {
 
+	/**
+	 * Called when state machine joined an ensemble.
+	 *
+	 * @param context the state machine context
+	 */
 	void stateMachineJoined(StateMachineContext<S, E> context);
 
+	/**
+	 * Called when state machine left an ensemble.
+	 *
+	 * @param context the state machine context
+	 */
 	void stateMachineLeft(StateMachineContext<S, E> context);
 
+	/**
+	 * Called when ensemble is discovering a state change.
+	 *
+	 * @param context the state machine context
+	 */
 	void stateChanged(StateMachineContext<S, E> context);
 
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java

@@ -64,4 +64,6 @@ public interface StateMachineEnsemble<S, E> {
 	 */
 	void setState(StateMachineContext<S, E> context);
 
+	StateMachineContext<S, E> getState();
+
 }

spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.statemachine.ensemble;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.StateMachineContext;
 import org.springframework.statemachine.support.LifecycleObjectSupport;
@@ -29,6 +31,8 @@ import org.springframework.statemachine.support.LifecycleObjectSupport;
  */
 public abstract class StateMachineEnsembleObjectSupport<S, E> extends LifecycleObjectSupport implements StateMachineEnsemble<S, E> {
 
+	private final static Log log = LogFactory.getLog(StateMachineEnsembleObjectSupport.class);
+
 	private final CompositeEnsembleListener<S, E> ensembleListener = new CompositeEnsembleListener<S, E>();
 
 	@Override
@@ -56,6 +60,9 @@ public abstract class StateMachineEnsembleObjectSupport<S, E> extends LifecycleO
 	}
 
 	protected void notifyStateChanged(StateMachineContext<S, E> context) {
+		if (log.isTraceEnabled()) {
+			log.trace("Notify notifyStateChanged " + context);
+		}
 		ensembleListener.stateChanged(context);
 	}
 

spring-statemachine-core/src/main/java/org/springframework/statemachine/region/Region.java

@@ -33,6 +33,13 @@ import org.springframework.statemachine.transition.Transition;
  */
 public interface Region<S, E> {
 
+	/**
+	 * Gets the region and state machine unique id.
+	 *
+	 * @return the region and state machine unique id
+	 */
+	String getId();
+
 	/**
 	 * Start the region.
 	 */

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

@@ -23,6 +23,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.Map.Entry;
 
 import org.apache.commons.logging.Log;
@@ -107,6 +108,8 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
 
 	private Boolean initialEnabled = null;
 
+	private String id = UUID.randomUUID().toString();
+
 	/**
 	 * Instantiates a new abstract state machine.
 	 *
@@ -445,8 +448,14 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
 	}
 
 	@Override
-	public void addStateChangeInterceptor(StateChangeInterceptor<S, E> interceptor) {
-		getStateChangeInterceptors().add(interceptor);
+	public void addStateMachineInterceptor(StateMachineInterceptor<S, E> interceptor) {
+		getStateMachineInterceptors().add(interceptor);
+		stateMachineExecutor.addStateMachineInterceptor(interceptor);
+	}
+
+	@Override
+	public String getId() {
+		return id;
 	}
 
 	protected boolean acceptEvent(Message<E> message) {
@@ -482,22 +491,29 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
 		return false;
 	}
 
-	private boolean callStateChangeInterceptors(State<S,E> state, Message<E> message, Transition<S,E> transition, StateMachine<S, E> stateMachine) {
+	private boolean callPreStateChangeInterceptors(State<S,E> state, Message<E> message, Transition<S,E> transition, StateMachine<S, E> stateMachine) {
 		try {
-			getStateChangeInterceptors().preStateChange(state, message, transition, stateMachine);
+			getStateMachineInterceptors().preStateChange(state, message, transition, stateMachine);
 		} catch (Exception e) {
-			log.info("Interceptors threw and exception, skipping state change", e);
+			log.info("Interceptors threw exception, skipping state change", e);
 			return false;
 		}
 		return true;
 	}
 
+	private void callPostStateChangeInterceptors(State<S,E> state, Message<E> message, Transition<S,E> transition, StateMachine<S, E> stateMachine) {
+		try {
+			getStateMachineInterceptors().postStateChange(state, message, transition, stateMachine);
+		} catch (Exception e) {
+		}
+	}
+
 	private boolean isInitialTransition(Transition<S,E> transition) {
 		return transition != null && transition.getKind() == TransitionKind.INITIAL;
 	}
 
 	private void switchToState(State<S,E> state, Message<E> message, Transition<S,E> transition, StateMachine<S, E> stateMachine) {
-		if (!isInitialTransition(transition) && !callStateChangeInterceptors(state, message, transition, stateMachine)) {
+		if (!isInitialTransition(transition) && !callPreStateChangeInterceptors(state, message, transition, stateMachine)) {
 			return;
 		}
 		// TODO: need to make below more clear when
@@ -517,6 +533,8 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
 			setCurrentState(state, message, transition, true, stateMachine);
 		}
 
+		callPostStateChangeInterceptors(state, message, transition, stateMachine);
+
 		stateMachineExecutor.execute();
 		if (isComplete()) {
 			stop();

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

@@ -65,4 +65,9 @@ public class DefaultExtendedState implements ExtendedState {
 		return (T) value;
 	}
 
+	@Override
+	public String toString() {
+		return "DefaultExtendedState [variables=" + variables + "]";
+	}
+
 }

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

@@ -37,6 +37,7 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.statemachine.ExtendedState;
 import org.springframework.statemachine.StateContext;
 import org.springframework.statemachine.StateMachine;
+import org.springframework.statemachine.StateMachineSystemConstants;
 import org.springframework.statemachine.state.State;
 import org.springframework.statemachine.transition.Transition;
 import org.springframework.statemachine.trigger.DefaultTriggerContext;
@@ -86,6 +87,9 @@ public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport im
 
 	private StateMachineExecutorTransit<S, E> stateMachineExecutorTransit;
 
+    private final StateMachineInterceptorList<S, E> interceptors =
+            new StateMachineInterceptorList<S, E>();
+
 	/**
 	 * Instantiates a new default state machine executor.
 	 *
@@ -154,6 +158,11 @@ public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport im
 		initialHandled.set(!enabled);
 	}
 
+	@Override
+	public void addStateMachineInterceptor(StateMachineInterceptor<S, E> interceptor) {
+		interceptors.add(interceptor);
+	}
+
 	private void handleTriggerTrans(List<Transition<S, E>> trans, Message<E> queuedMessage) {
 		for (Transition<S, E> t : trans) {
 			if (t == null) {
@@ -167,10 +176,17 @@ public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport im
 			if (!StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) {
 				continue;
 			}
+
 			StateContext<S, E> stateContext = buildStateContext(queuedMessage, t, relayStateMachine);
+			stateContext = interceptors.preTransition(stateContext);
+			if (stateContext == null) {
+				break;
+			}
+
 			boolean transit = t.transit(stateContext);
 			if (transit) {
 				stateMachineExecutorTransit.transit(t, stateContext, queuedMessage);
+				interceptors.postTransition(stateContext);
 				break;
 			}
 		}
@@ -319,9 +335,17 @@ public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport im
 
 	private StateContext<S, E> buildStateContext(Message<E> message, Transition<S,E> transition, StateMachine<S, E> stateMachine) {
 		E event = message != null ? message.getPayload() : null;
+
+		// TODO: maybe a direct use of MessageHeaders is wring, combine
+		//       payload and headers as a message?
+
+		// add sm id to headers so that user of a StateContext can
+		// see who initiated this transition
 		MessageHeaders messageHeaders = message != null ? message.getHeaders() : new MessageHeaders(
 				new HashMap<String, Object>());
-		return new DefaultStateContext<S, E>(event, messageHeaders, extendedState, transition, stateMachine);
+		Map<String, Object> map = new HashMap<String, Object>(messageHeaders);
+		map.put(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER, stateMachine.getId());
+		return new DefaultStateContext<S, E>(event, new MessageHeaders(map), extendedState, transition, stateMachine);
 	}
 
 	private void registerTriggerListener() {

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

@@ -86,6 +86,13 @@ public interface StateMachineExecutor<S, E> {
 	 */
 	void setStateMachineExecutorTransit(StateMachineExecutorTransit<S, E> stateMachineExecutorTransit);
 
+	/**
+	 * Adds the state machine interceptor.
+	 *
+	 * @param interceptor the interceptor
+	 */
+	void addStateMachineInterceptor(StateMachineInterceptor<S, E> interceptor);
+
 	/**
 	 * Callback interface when executor wants to handle transit.
 	 */

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

@@ -0,0 +1,76 @@
+/*
+ * 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.state.State;
+import org.springframework.statemachine.transition.Transition;
+
+/**
+ * Interface which can be registered with a state machine and can be used
+ * to intercept and break a state change chain.
+ *
+ * @author Janne Valkealahti
+ *
+ * @param <S> the type of state
+ * @param <E> the type of event
+ */
+public interface StateMachineInterceptor<S, E> {
+
+	/**
+	 * Called prior of a state change. Throwing an exception
+	 * from this method will stop a state change logic.
+	 *
+	 * @param state the state
+	 * @param message the message
+	 * @param transition the transition
+	 * @param stateMachine the state machine
+	 */
+	void preStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
+			StateMachine<S, E> stateMachine);
+
+	/**
+	 * Called after a state change.
+	 *
+	 * @param state the state
+	 * @param message the message
+	 * @param transition the transition
+	 * @param stateMachine the state machine
+	 */
+	void postStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
+			StateMachine<S, E> stateMachine);
+
+	/**
+	 * Called prior of a start of a transition. Returning
+	 * {@code null} from this method will break the transtion
+	 * chain.
+	 *
+	 * @param stateContext the state context
+	 * @return the state context
+	 */
+	StateContext<S, E> preTransition(StateContext<S, E> stateContext);
+
+	/**
+	 * Called after of a transition if transition happened.
+	 *
+	 * @param stateContext the state context
+	 * @return the state context
+	 */
+	StateContext<S, E> postTransition(StateContext<S, E> stateContext);
+
+}

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

@@ -16,12 +16,39 @@
 package org.springframework.statemachine.support;
 
 import org.springframework.messaging.Message;
+import org.springframework.statemachine.StateContext;
 import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.state.State;
 import org.springframework.statemachine.transition.Transition;
 
-public interface StateChangeInterceptor<S, E> {
+/**
+ * Adapter helper implementation for {@link StateMachineInterceptor}.
+ *
+ * @author Janne Valkealahti
+ *
+ * @param <S> the type of state
+ * @param <E> the type of event
+ */
+public class StateMachineInterceptorAdapter<S, E> implements StateMachineInterceptor<S, E> {
 
-	void preStateChange(State<S,E> state, Message<E> message, Transition<S,E> transition, StateMachine<S, E> stateMachine);
+	@Override
+	public void preStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
+			StateMachine<S, E> stateMachine) {
+	}
+
+	@Override
+	public void postStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
+			StateMachine<S, E> stateMachine) {
+	}
+
+	@Override
+	public StateContext<S, E> preTransition(StateContext<S, E> stateContext) {
+		return stateContext;
+	}
+
+	@Override
+	public StateContext<S, E> postTransition(StateContext<S, E> stateContext) {
+		return stateContext;
+	}
 
 }

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

@@ -0,0 +1,134 @@
+/*
+ * 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.Collection;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.springframework.messaging.Message;
+import org.springframework.statemachine.StateContext;
+import org.springframework.statemachine.StateMachine;
+import org.springframework.statemachine.state.State;
+import org.springframework.statemachine.transition.Transition;
+
+/**
+ * Support class working with a {@link StateMachineInterceptor}s.
+ *
+ * @author Janne Valkealahti
+ *
+ * @param <S> the type of state
+ * @param <E> the type of event
+ */
+public class StateMachineInterceptorList<S, E> {
+
+	private final List<StateMachineInterceptor<S, E>> interceptors = new CopyOnWriteArrayList<StateMachineInterceptor<S, E>>();
+
+	/**
+	 * Sets the interceptors, clears any existing interceptors.
+	 *
+	 * @param interceptors the list of interceptors
+	 * @return <tt>true</tt> if interceptor list changed as a result of the
+	 *         call
+	 */
+	public boolean set(List<StateMachineInterceptor<S, E>> interceptors) {
+		synchronized (interceptors) {
+			interceptors.clear();
+			return interceptors.addAll(interceptors);
+		}
+	}
+
+	/**
+	 * Adds interceptor to the list.
+	 *
+	 * @param interceptor the interceptor
+	 * @return <tt>true</tt> (as specified by {@link Collection#add})
+	 */
+	public boolean add(StateMachineInterceptor<S, E> interceptor) {
+		return interceptors.add(interceptor);
+	}
+
+	/**
+	 * Removes interceptor from the list.
+	 *
+	 * @param interceptor the interceptor
+	 * @return <tt>true</tt> (as specified by {@link Collection#remove})
+	 */
+	public boolean remove(StateMachineInterceptor<S, E> interceptor) {
+		return interceptors.remove(interceptor);
+	}
+
+	/**
+	 * Pre state change.
+	 *
+	 * @param state the state
+	 * @param message the message
+	 * @param transition the transition
+	 * @param stateMachine the state machine
+	 */
+	public void preStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
+			StateMachine<S, E> stateMachine) {
+		for (StateMachineInterceptor<S, E> interceptor : interceptors) {
+			interceptor.preStateChange(state, message, transition, stateMachine);
+		}
+	}
+
+	/**
+	 * Post state change.
+	 *
+	 * @param state the state
+	 * @param message the message
+	 * @param transition the transition
+	 * @param stateMachine the state machine
+	 */
+	public void postStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
+			StateMachine<S, E> stateMachine) {
+		for (StateMachineInterceptor<S, E> interceptor : interceptors) {
+			interceptor.postStateChange(state, message, transition, stateMachine);
+		}
+	}
+
+	/**
+	 * Pre transition.
+	 *
+	 * @param stateContext the state context
+	 * @return the state context
+	 */
+	public StateContext<S, E> preTransition(StateContext<S, E> stateContext) {
+		for (StateMachineInterceptor<S, E> interceptor : interceptors) {
+			if ((stateContext = interceptor.preTransition(stateContext)) == null) {
+				break;
+			}
+		}
+		return stateContext;
+	}
+
+	/**
+	 * Post transition.
+	 *
+	 * @param stateContext the state context
+	 * @return the state context
+	 */
+	public StateContext<S, E> postTransition(StateContext<S, E> stateContext) {
+		for (StateMachineInterceptor<S, E> interceptor : interceptors) {
+			if ((stateContext = interceptor.postTransition(stateContext)) == null) {
+				break;
+			}
+		}
+		return stateContext;
+	}
+
+}

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

@@ -15,10 +15,8 @@
  */
 package org.springframework.statemachine.support;
 
-import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -52,8 +50,8 @@ public abstract class StateMachineObjectSupport<S, E> extends LifecycleObjectSup
 	/** Flag for application context events */
 	private boolean contextEventsEnabled = true;
 
-    private final StateChangeInterceptorList interceptors =
-            new StateChangeInterceptorList();
+    private final StateMachineInterceptorList<S, E> interceptors =
+            new StateMachineInterceptorList<S, E>();
 
 	/**
 	 * Gets the state machine event publisher.
@@ -195,11 +193,11 @@ public abstract class StateMachineObjectSupport<S, E> extends LifecycleObjectSup
 		//       re-scheduling is needed.
 	}
 
-	protected StateChangeInterceptorList getStateChangeInterceptors() {
+	protected StateMachineInterceptorList<S, E> getStateMachineInterceptors() {
 		return interceptors;
 	}
 
-	protected void setStateChangeInterceptors(List<StateChangeInterceptor<S,E>> interceptors) {
+	protected void setStateMachineInterceptors(List<StateMachineInterceptor<S,E>> interceptors) {
 		Collections.sort(interceptors, new OrderComparator());
 	    this.interceptors.set(interceptors);
 	}
@@ -259,53 +257,4 @@ public abstract class StateMachineObjectSupport<S, E> extends LifecycleObjectSup
 
 	}
 
-	protected class StateChangeInterceptorList {
-
-		private final List<StateChangeInterceptor<S, E>> interceptors = new CopyOnWriteArrayList<StateChangeInterceptor<S, E>>();
-
-		/**
-		 * Sets the interceptors, clears any existing interceptors.
-		 *
-		 * @param interceptors the list of interceptors
-		 * @return <tt>true</tt> if interceptor list changed as a result of the
-		 *         call
-		 */
-		public boolean set(List<StateChangeInterceptor<S, E>> interceptors) {
-			synchronized (interceptors) {
-				interceptors.clear();
-				return interceptors.addAll(interceptors);
-			}
-		}
-
-		/**
-		 * Adds interceptor to the list.
-		 *
-		 * @param interceptor the interceptor
-		 * @return <tt>true</tt> (as specified by {@link Collection#add})
-		 */
-		public boolean add(StateChangeInterceptor<S, E> interceptor) {
-			return interceptors.add(interceptor);
-		}
-
-		/**
-		 * Removes interceptor from the list.
-		 *
-		 * @param interceptor the interceptor
-		 * @return <tt>true</tt> (as specified by {@link Collection#remove})
-		 */
-		public boolean remove(StateChangeInterceptor<S, E> interceptor) {
-			return interceptors.remove(interceptor);
-		}
-
-		/**
-		 * Handles the pre state change calls.
-		 */
-		void preStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
-				StateMachine<S, E> stateMachine) {
-			for (StateChangeInterceptor<S, E> interceptor : interceptors) {
-				interceptor.preStateChange(state, message, transition, stateMachine);
-			}
-		}
-
-	}
 }

spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java

@@ -29,7 +29,7 @@ import org.springframework.statemachine.StateMachine;
 import org.springframework.statemachine.StateMachineContext;
 import org.springframework.statemachine.listener.StateMachineListener;
 import org.springframework.statemachine.state.State;
-import org.springframework.statemachine.support.StateChangeInterceptor;
+import org.springframework.statemachine.support.StateMachineInterceptor;
 import org.springframework.statemachine.transition.Transition;
 
 public class StateMachineAccessTests {
@@ -92,7 +92,7 @@ public class StateMachineAccessTests {
 		}
 
 		@Override
-		public void addStateChangeInterceptor(StateChangeInterceptor<String, String> interceptor) {
+		public void addStateMachineInterceptor(StateMachineInterceptor<String, String> interceptor) {
 		}
 
 		@Override
@@ -164,6 +164,11 @@ public class StateMachineAccessTests {
 		public void setInitialEnabled(boolean enabled) {
 		}
 
+		@Override
+		public String getId() {
+			return null;
+		}
+
 	}
 
 }

spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java

@@ -48,4 +48,9 @@ public class InMemoryStateMachineEnsemble<S, E> extends StateMachineEnsembleObje
 		notifyStateChanged(context);
 	}
 
+	@Override
+	public StateMachineContext<S, E> getState() {
+		return current;
+	}
+
 }

spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateChangeInterceptorTests.java

@@ -65,7 +65,7 @@ public class StateChangeInterceptorTests extends AbstractStateMachineTests {
 
 			@Override
 			public void apply(StateMachineAccess<States, Events> function) {
-				function.addStateChangeInterceptor(interceptor);
+				function.addStateMachineInterceptor(interceptor);
 			}
 		});
 
@@ -74,10 +74,8 @@ public class StateChangeInterceptorTests extends AbstractStateMachineTests {
 		assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true));
 		assertThat(listener.stateChangedCount, is(3));
 		assertThat(machine.getState().getIds(), containsInAnyOrder(States.S0, States.S1, States.S11));
-
 		assertThat((Integer)machine.getExtendedState().getVariables().get("foo"), is(0));
 
-
 		listener.reset(3);
 		interceptor.reset(1);
 		machine.sendEvent(Events.C);
@@ -86,6 +84,9 @@ public class StateChangeInterceptorTests extends AbstractStateMachineTests {
 		assertThat(interceptor.preStateChangeLatch.await(2, TimeUnit.SECONDS), is(true));
 		assertThat(interceptor.preStateChangeCount, is(1));
 		assertThat(machine.getState().getIds(), containsInAnyOrder(States.S0, States.S2, States.S21, States.S211));
+		machine.sendEvent(Events.H);
+		assertThat(machine.getState().getIds(), containsInAnyOrder(States.S0, States.S2, States.S21, States.S211));
+		assertThat((Integer)machine.getExtendedState().getVariables().get("foo"), is(1));
 	}
 
 	@Configuration
@@ -266,7 +267,7 @@ public class StateChangeInterceptorTests extends AbstractStateMachineTests {
 
 	}
 
-	private static class TestStateChangeInterceptor implements StateChangeInterceptor<States, Events> {
+	private static class TestStateChangeInterceptor implements StateMachineInterceptor<States, Events> {
 
 		volatile CountDownLatch preStateChangeLatch = new CountDownLatch(1);
 		volatile int preStateChangeCount = 0;
@@ -279,6 +280,22 @@ public class StateChangeInterceptorTests extends AbstractStateMachineTests {
 
 		}
 
+		@Override
+		public void postStateChange(State<States, Events> state, Message<Events> message,
+				Transition<States, Events> transition, StateMachine<States, Events> stateMachine) {
+		}
+
+		@Override
+		public StateContext<States, Events> preTransition(StateContext<States, Events> stateContext) {
+			return stateContext;
+		}
+
+		@Override
+		public StateContext<States, Events> postTransition(StateContext<States, Events> stateContext) {
+			return stateContext;
+		}
+
+
 		public void reset(int c1) {
 			preStateChangeLatch = new CountDownLatch(c1);
 			preStateChangeCount = 0;

spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateContextExpressionMethodsTests.java

@@ -194,6 +194,11 @@ public class StateContextExpressionMethodsTests {
 			return null;
 		}
 
+		@Override
+		public String getId() {
+			return null;
+		}
+
 	}
 
 }

spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java

@@ -26,7 +26,7 @@ import org.springframework.statemachine.listener.AbstractCompositeListener;
 import org.springframework.statemachine.state.State;
 import org.springframework.statemachine.support.DefaultStateMachineContext;
 import org.springframework.statemachine.support.LifecycleObjectSupport;
-import org.springframework.statemachine.support.StateChangeInterceptor;
+import org.springframework.statemachine.support.StateMachineInterceptorAdapter;
 import org.springframework.statemachine.transition.Transition;
 import org.springframework.util.Assert;
 
@@ -59,7 +59,7 @@ public class PersistStateMachineHandler extends LifecycleObjectSupport {
 
 			@Override
 			public void apply(StateMachineAccess<String, String> function) {
-				function.addStateChangeInterceptor(interceptor);
+				function.addStateMachineInterceptor(interceptor);
 			}
 		});
 	}
@@ -108,17 +108,17 @@ public class PersistStateMachineHandler extends LifecycleObjectSupport {
 		 * @param transition the transition
 		 * @param stateMachine the state machine
 		 */
-		void onPersist(State<String, String> state, Message<String> message, Transition<String, String> transition, StateMachine<String, String> stateMachine);
+		void onPersist(State<String, String> state, Message<String> message, Transition<String, String> transition,
+				StateMachine<String, String> stateMachine);
 	}
 
-	private class PersistingStateChangeInterceptor implements StateChangeInterceptor<String, String> {
+	private class PersistingStateChangeInterceptor extends StateMachineInterceptorAdapter<String, String> {
 
 		@Override
 		public void preStateChange(State<String, String> state, Message<String> message,
 				Transition<String, String> transition, StateMachine<String, String> stateMachine) {
 			listeners.onPersist(state, message, transition, stateMachine);
 		}
-
 	}
 
 	private class CompositePersistStateChangeListener extends AbstractCompositeListener<PersistStateChangeListener> implements

spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java

@@ -66,6 +66,7 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 	private final boolean cleanState;
 	private final StateMachinePersist<S, E, Stat> persist;
 	private final AtomicReference<StateWrapper> stateRef = new AtomicReference<StateWrapper>();
+	private final AtomicReference<StateWrapper> notifyRef = new AtomicReference<StateWrapper>();
 	private final CuratorWatcher watcher = new StateWatcher();
 	private PersistentEphemeralNode node;
 
@@ -97,6 +98,7 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 		this.memberPath = basePath + "/" + PATH_MEMBERS;
 		this.mutexPath = basePath + "/" + PATH_MUTEX;
 		this.persist = new ZookeeperStateMachinePersist<S, E>(curatorClient, statePath, logPath, logSize);
+		setAutoStartup(true);
 	}
 
 	@Override
@@ -106,6 +108,20 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 
 	@Override
 	protected void doStart() {
+		// initially setting a watcher here, further watchers
+		// will be set when events are received.
+		registerWatcherForStatePath();
+
+		StateWrapper stateWrapper = stateRef.get();
+		if (stateWrapper == null) {
+			try {
+				StateWrapper currentStateWrapper = readCurrentContext();
+				stateRef.set(new StateWrapper(currentStateWrapper.context, currentStateWrapper.version));
+				stateWrapper = stateRef.get();
+			} catch (Exception e) {
+				log.error("Error reading current state during start", e);
+			}
+		}
 	}
 
 	@Override
@@ -123,15 +139,6 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 	@Override
 	public void join(StateMachine<S, E> stateMachine) {
 		StateWrapper stateWrapper = stateRef.get();
-		if (stateWrapper == null) {
-			try {
-				StateWrapper currentStateWrapper = readCurrentContext();
-				stateRef.set(new StateWrapper(currentStateWrapper.context, currentStateWrapper.version));
-				stateWrapper = stateRef.get();
-			} catch (Exception e) {
-				log.error("Error reading current state during join", e);
-			}
-		}
 		notifyJoined(stateWrapper != null ? stateWrapper.context : null);
 	}
 
@@ -148,7 +155,10 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 	}
 
 	@Override
-	public void setState(StateMachineContext<S, E> context) {
+	public synchronized void setState(StateMachineContext<S, E> context) {
+		if (log.isDebugEnabled()) {
+			log.debug("Setting state context=" + context);
+		}
 		try {
 			Stat stat = new Stat();
 			StateWrapper stateWrapper = stateRef.get();
@@ -162,12 +172,15 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 		}
 	}
 
+	@Override
+	public StateMachineContext<S, E> getState() {
+		return readCurrentContext().context;
+	}
+
 	private StateWrapper readCurrentContext() {
 		try {
 			Stat stat = new Stat();
-
-			// TODO: not nice that we need to set watcher here when persister is reading data
-			curatorClient.getData().usingWatcher(watcher).forPath(statePath);
+			registerWatcherForStatePath();
 			StateMachineContext<S, E> context = persist.read(stat);
 			return new StateWrapper(context, stat.getVersion());
 		} catch (Exception e) {
@@ -175,8 +188,13 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 		}
 	}
 
+	/**
+	 * Create all needed paths including what ZookeeperStateMachinePersist
+	 * is going to need because it doesn't handle any path creation. We also
+	 * use a mutex lock to make a decision if cleanState is enabled to wipe
+	 * out existing data.
+	 */
 	private void initPaths() {
-
 		InterProcessSemaphoreMutex mutex = new InterProcessSemaphoreMutex(curatorClient, mutexPath);
 		try {
 			if (log.isTraceEnabled()) {
@@ -224,8 +242,48 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 		}
 	}
 
+	/**
+	 * Register existing {@link CuratorWatcher} for a state path.
+	 */
+	private void registerWatcherForStatePath() {
+		try {
+			if (curatorClient.getState() != CuratorFrameworkState.STOPPED) {
+				curatorClient.checkExists().usingWatcher(watcher).forPath(statePath);
+			}
+		} catch (Exception e) {
+			log.warn("Registering wacher for path " + statePath + " threw error", e);
+		}
+	}
+
+	private void mayNotifyStateChanged(StateWrapper wrapper) {
+		StateWrapper notifyWrapper = notifyRef.get();
+		if (notifyWrapper == null) {
+			notifyRef.set(wrapper);
+			notifyStateChanged(wrapper.context);
+		} else if (wrapper.version > notifyWrapper.version) {
+			notifyRef.set(wrapper);
+			notifyStateChanged(wrapper.context);
+		}
+	}
+
+	private void traceLogWrappers(StateWrapper currentWrapper, StateWrapper notifyWrapper, StateWrapper newWrapper) {
+		if (log.isTraceEnabled()) {
+			log.trace("Wrappers \ncurrentWrapper=[" + currentWrapper + "] \nnotifyWrapper=[" + notifyWrapper
+					+ "] \nnewWrapper=[" + newWrapper + "]");
+		}
+	}
+
 	private class StateWatcher implements CuratorWatcher {
 
+		// zk is not really reliable for watching events because
+		// you need to re-register watcher when it fires. most likely
+		// we will miss events so need to do little tricks here via
+		// event logs.
+
+		// NOTE: because paths are pre-created, version always start
+		//       from 1 when real data is set. initial path contains
+		//       empty data with version 0.
+
 		@Override
 		public void process(WatchedEvent event) throws Exception {
 			if (log.isTraceEnabled()) {
@@ -233,26 +291,44 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 			}
 			switch (event.getType()) {
 			case NodeDataChanged:
-				StateWrapper currentStateWrapper = stateRef.get();
-				StateWrapper newStateWrapper = readCurrentContext();
-				if (log.isTraceEnabled()) {
-					log.trace("NodeDataChanged currentStateWrapper=" + currentStateWrapper + " newStateWrapper=" + newStateWrapper);
-				}
+				try {
+					StateWrapper currentWrapper = stateRef.get();
+					StateWrapper notifyWrapper = notifyRef.get();
+					StateWrapper newWrapper = readCurrentContext();
+					traceLogWrappers(currentWrapper, notifyWrapper, newWrapper);
 
-				// if we don't have a previous version, we've missed an update.
-				// we're going to replay those from log paths.
-				if (currentStateWrapper.version + 1 == newStateWrapper.version
-						&& stateRef.compareAndSet(currentStateWrapper, newStateWrapper)) {
-					if (log.isTraceEnabled()) {
-						log.trace("Notify state change with new context");
+					if (currentWrapper.version + 1 == newWrapper.version
+							&& stateRef.compareAndSet(currentWrapper, newWrapper)) {
+						mayNotifyStateChanged(newWrapper);
+					} else {
+						final int start = (notifyWrapper != null ? (notifyWrapper.version) : 0) % logSize;
+						int count = newWrapper.version - (notifyWrapper != null ? (notifyWrapper.version) : 0);
+						if (log.isDebugEnabled()) {
+							log.debug("Events missed, trying to replay start " + start + " count " + count);
+						}
+						for (int i = start; i < (start + count); i++) {
+							try {
+								Stat stat = new Stat();
+								StateMachineContext<S, E> context = ((ZookeeperStateMachinePersist<S, E>) persist)
+										.readLog(i, stat);
+								int ver = (stat.getVersion() - 1) * logSize + (i + 1);
+								if (log.isDebugEnabled()) {
+									log.debug("Replay position " + i + " with version " + ver);
+								}
+								StateWrapper wrapper = new StateWrapper(context, ver);
+								mayNotifyStateChanged(wrapper);
+							} catch (Exception e) {
+								log.error("error reading log", e);
+							}
+						}
 					}
-					notifyStateChanged(newStateWrapper.context);
-				} else {
-
+				} catch (Exception e) {
+					log.error("Error handling event", e);
 				}
+				registerWatcherForStatePath();
 				break;
 			default:
-				curatorClient.checkExists().usingWatcher(this).forPath(statePath);
+				registerWatcherForStatePath();
 				break;
 			}
 		}
@@ -260,7 +336,8 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 	}
 
 	/**
-	 * Wrapper object for a {@link StateMachineContext}.
+	 * Wrapper object for a {@link StateMachineContext} and its
+	 * current version.
 	 */
 	private class StateWrapper {
 		private final StateMachineContext<S, E> context;

spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java

@@ -121,9 +121,11 @@ public class ZookeeperStateMachinePersist<S, E> implements StateMachinePersist<S
 
 	@Override
 	public StateMachineContext<S, E> read(Stat stat) throws Exception {
-		byte[] data = curatorClient.getData().storingStatIn(stat).forPath(path);
-		StateMachineContext<S, E> context = deserialize(data);
-		return context;
+		return deserialize(curatorClient.getData().storingStatIn(stat).forPath(path));
+	}
+
+	public StateMachineContext<S, E> readLog(int version, Stat stat) throws Exception {
+		return deserialize(curatorClient.getData().storingStatIn(stat).forPath(logPath + "/" + version));
 	}
 
 	private byte[] serialize(StateMachineContext<S, E> context) {

spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java

@@ -18,10 +18,13 @@ package org.springframework.statemachine.zookeeper;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
 import static org.junit.Assert.assertThat;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -139,7 +142,123 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests {
 		assertThat(curatorClient.getData().forPath("/foo/data/log").length, is(0));
 	}
 
-	//
+	@Test
+	public void testLogs() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		ZookeeperStateMachineEnsemble<String, String> ensemble =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo", true, 4);
+
+		ensemble.afterPropertiesSet();
+		ensemble.start();
+
+		assertThat(curatorClient.checkExists().forPath("/foo/data/log"), notNullValue());
+		assertThat(curatorClient.checkExists().forPath("/foo/data/log/0"), notNullValue());
+		assertThat(curatorClient.checkExists().forPath("/foo/data/log/1"), notNullValue());
+		assertThat(curatorClient.checkExists().forPath("/foo/data/log/2"), notNullValue());
+		assertThat(curatorClient.checkExists().forPath("/foo/data/log/3"), notNullValue());
+		assertThat(curatorClient.checkExists().forPath("/foo/data/log/4"), nullValue());
+		assertThat(curatorClient.getData().forPath("/foo/data/log/0").length, is(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/1").length, is(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/2").length, is(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/3").length, is(0));
+
+		ensemble.setState(new DefaultStateMachineContext<String, String>("S1","E1", new HashMap<String, Object>(), new DefaultExtendedState()));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/0").length, greaterThan(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/1").length, is(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/2").length, is(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/3").length, is(0));
+
+		ensemble.setState(new DefaultStateMachineContext<String, String>("S2","E1", new HashMap<String, Object>(), new DefaultExtendedState()));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/0").length, greaterThan(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/1").length, greaterThan(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/2").length, is(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/3").length, is(0));
+
+		ensemble.setState(new DefaultStateMachineContext<String, String>("S3","E1", new HashMap<String, Object>(), new DefaultExtendedState()));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/0").length, greaterThan(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/1").length, greaterThan(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/2").length, greaterThan(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/3").length, is(0));
+
+		ensemble.setState(new DefaultStateMachineContext<String, String>("S4","E1", new HashMap<String, Object>(), new DefaultExtendedState()));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/0").length, greaterThan(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/1").length, greaterThan(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/2").length, greaterThan(0));
+		assertThat(curatorClient.getData().forPath("/foo/data/log/3").length, greaterThan(0));
+	}
+
+	@Test(expected = IllegalStateException.class)
+	public void testIllegalLogSize() throws Exception {
+				new ZookeeperStateMachineEnsemble<String, String>(null, "/foo", true, 3);
+	}
+
+	@Test
+	public void testContextEventsNotMissedBurstNoOverflow() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		ZookeeperStateMachineEnsemble<String, String> ensemble =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+
+		TestEnsembleListener listener = new TestEnsembleListener();
+		ensemble.addEnsembleListener(listener);
+
+		ensemble.afterPropertiesSet();
+		ensemble.start();
+		listener.reset(0, 10);
+
+		for (int i = 0; i < 10; i++) {
+			ensemble.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+					new HashMap<String, Object>(), new DefaultExtendedState()));
+		}
+
+		assertThat(listener.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener.events.size(), is(10));
+
+		for (int i = 0; i < 10; i++) {
+			assertThat(listener.events.get(i).getEvent(), is("E" + i));
+		}
+	}
+
+	@Test
+	public void testContextEventsNotMissedSlowNoOverflow() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+
+		ZookeeperStateMachineEnsemble<String, String> ensemble =
+				new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
+
+		TestEnsembleListener listener = new TestEnsembleListener();
+		ensemble.addEnsembleListener(listener);
+
+		ensemble.afterPropertiesSet();
+		ensemble.start();
+		listener.reset(0, 10);
+
+		for (int i = 0; i < 10; i++) {
+			ensemble.setState(new DefaultStateMachineContext<String, String>("S" + i, "E" + i,
+					new HashMap<String, Object>(), new DefaultExtendedState()));
+			Thread.sleep(500);
+		}
+
+		assertThat(listener.eventLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(listener.events.size(), is(10));
+
+		for (int i = 0; i < 10; i++) {
+			assertThat(listener.events.get(i).getEvent(), is("E" + i));
+		}
+	}
 
 	@Override
 	protected AnnotationConfigApplicationContext buildContext() {
@@ -150,6 +269,7 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests {
 
 		volatile CountDownLatch joinedLatch = new CountDownLatch(1);
 		volatile CountDownLatch eventLatch = new CountDownLatch(1);
+		volatile List<StateMachineContext<String, String>> events = new ArrayList<StateMachineContext<String,String>>();
 
 		@Override
 		public void stateMachineJoined(StateMachineContext<String, String> context) {
@@ -162,9 +282,16 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests {
 
 		@Override
 		public void stateChanged(StateMachineContext<String, String> context) {
+			events.add(context);
 			eventLatch.countDown();
 		}
 
+		public void reset(int c1, int c2) {
+			joinedLatch = new CountDownLatch(c1);
+			eventLatch = new CountDownLatch(c2);
+			events.clear();
+		}
+
 	}
 
 	private class TestStateMachine implements StateMachine<String, String> {
@@ -230,6 +357,11 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests {
 			return null;
 		}
 
+		@Override
+		public String getId() {
+			return null;
+		}
+
 	}
 
 }

spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java

@@ -60,6 +60,37 @@ public class ZookeeperStateMachinePersistTests extends AbstractZookeeperTests {
 		assertThat(contextOut.getEvent(), is(contextIn.getEvent()));
 	}
 
+	@Test
+	public void testLogs() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+		curatorClient.create().forPath("/KryoStateMachinePersistTests");
+
+		ZookeeperStateMachinePersist<String, String> persist = new ZookeeperStateMachinePersist<String, String>(
+				curatorClient, "/KryoStateMachinePersistTests", "/KryoStateMachinePersistTestsLogs", 32);
+
+		for (int i = 0; i < 10; i++) {
+			curatorClient.create().creatingParentsIfNeeded().forPath("/KryoStateMachinePersistTestsLogs/" + i);
+		}
+
+		for (int i = 0; i < 10; i++) {
+			Stat stat = new Stat();
+			stat.setVersion(i);
+			StateMachineContext<String, String> contextOut =
+					new DefaultStateMachineContext<String, String>("S" + i, "E" + i, new HashMap<String, Object>(), new DefaultExtendedState());
+			persist.write(contextOut, stat);
+		}
+
+		for (int i = 0; i < 10; i++) {
+			StateMachineContext<String, String> contextIn = persist.readLog(i, new Stat());
+			assertThat(contextIn.getState(), is("S" + i));
+			assertThat(contextIn.getEvent(), is("E" + i));
+		}
+	}
+
 	@Test
 	public void testEventHeaders() throws Exception {
 		context.register(ZkServerConfig.class, BaseConfig.class);

spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java

@@ -269,7 +269,7 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests {
 		}
 	}
 
-	abstract static class SharedConfig2 extends SharedConfig1 {
+	public abstract static class SharedConfig2 extends SharedConfig1 {
 
 		@Override
 		public void configure(StateMachineConfigurationConfigurer<String, String> config) throws Exception {