From 0cdd8336484aadac7f0097993b925b4b0aaed886 Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Sun, 19 Jul 2015 10:33:28 +0100 Subject: [PATCH] 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. --- .../StateMachineSystemConstants.java | 5 +- .../access/StateMachineAccess.java | 6 +- .../ensemble/DistributedStateMachine.java | 50 +++++-- .../ensemble/EnsembleListeger.java | 15 ++ .../ensemble/StateMachineEnsemble.java | 2 + .../StateMachineEnsembleObjectSupport.java | 7 + .../statemachine/region/Region.java | 7 + .../support/AbstractStateMachine.java | 30 +++- .../support/DefaultExtendedState.java | 5 + .../support/DefaultStateMachineExecutor.java | 26 +++- .../support/StateMachineExecutor.java | 7 + .../support/StateMachineInterceptor.java | 76 ++++++++++ ...va => StateMachineInterceptorAdapter.java} | 31 +++- .../support/StateMachineInterceptorList.java | 134 +++++++++++++++++ .../support/StateMachineObjectSupport.java | 59 +------- .../access/StateMachineAccessTests.java | 9 +- .../InMemoryStateMachineEnsemble.java | 5 + .../support/StateChangeInterceptorTests.java | 25 +++- .../StateContextExpressionMethodsTests.java | 5 + .../persist/PersistStateMachineHandler.java | 10 +- .../ZookeeperStateMachineEnsemble.java | 137 ++++++++++++++---- .../ZookeeperStateMachinePersist.java | 8 +- .../ZookeeperStateMachineEnsembleTests.java | 134 ++++++++++++++++- .../ZookeeperStateMachinePersistTests.java | 31 ++++ .../zookeeper/ZookeeperStateMachineTests.java | 2 +- 25 files changed, 702 insertions(+), 124 deletions(-) create mode 100644 spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineInterceptor.java rename spring-statemachine-core/src/main/java/org/springframework/statemachine/support/{StateChangeInterceptor.java => StateMachineInterceptorAdapter.java} (50%) create mode 100644 spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineInterceptorList.java diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineSystemConstants.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineSystemConstants.java index 522c5b51..50edab51 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineSystemConstants.java +++ b/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_"; + } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java index 5d2342b1..acd39292 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java +++ b/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 { void resetStateMachine(StateMachineContext stateMachineContext); /** - * Adds the state change interceptor. + * Adds the state machine interceptor. * * @param interceptor the interceptor */ - void addStateChangeInterceptor(StateChangeInterceptor interceptor); + void addStateMachineInterceptor(StateMachineInterceptor interceptor); /** * Sets if initial state is enabled when a state machine is diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java index 1ba8ec60..452d0abe 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java +++ b/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 extends LifecycleObjectSupport implements StateMachine { private final static Log log = LogFactory.getLog(DistributedStateMachine.class); - private final String uuid = UUID.randomUUID().toString(); private final StateMachineEnsemble ensemble; private final StateMachine 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 extends LifecycleObjectSupport implem @Override public void apply(StateMachineAccess function) { - function.addStateChangeInterceptor(interceptor); + function.addStateMachineInterceptor(interceptor); } }); } @@ -99,7 +99,10 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem @Override public boolean sendEvent(Message 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 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 { + private class LocalStateMachineInterceptor implements StateMachineInterceptor { @Override public void preStateChange(State state, Message message, Transition transition, StateMachine stateMachine) { - if (message != null && ObjectUtils.nullSafeEquals(uuid, message.getHeaders().get("uuid"))) { - ensemble.setState(new DefaultStateMachineContext(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(transition.getTarget().getId(), message + .getPayload(), message.getHeaders(), stateMachine.getExtendedState())); } } + @Override + public void postStateChange(State state, Message message, Transition transition, + StateMachine stateMachine) { + } + + @Override + public StateContext preTransition(StateContext stateContext) { + return stateContext; + } + + @Override + public StateContext postTransition(StateContext stateContext) { + return stateContext; + } + } /** diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java index 00c5c8e5..cd8adf8b 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java @@ -27,10 +27,25 @@ import org.springframework.statemachine.StateMachineContext; */ public interface EnsembleListeger { + /** + * Called when state machine joined an ensemble. + * + * @param context the state machine context + */ void stateMachineJoined(StateMachineContext context); + /** + * Called when state machine left an ensemble. + * + * @param context the state machine context + */ void stateMachineLeft(StateMachineContext context); + /** + * Called when ensemble is discovering a state change. + * + * @param context the state machine context + */ void stateChanged(StateMachineContext context); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java index fbaf263f..913bae67 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java @@ -64,4 +64,6 @@ public interface StateMachineEnsemble { */ void setState(StateMachineContext context); + StateMachineContext getState(); + } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java index dec8cac4..cc4fa758 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java +++ b/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 extends LifecycleObjectSupport implements StateMachineEnsemble { + private final static Log log = LogFactory.getLog(StateMachineEnsembleObjectSupport.class); + private final CompositeEnsembleListener ensembleListener = new CompositeEnsembleListener(); @Override @@ -56,6 +60,9 @@ public abstract class StateMachineEnsembleObjectSupport extends LifecycleO } protected void notifyStateChanged(StateMachineContext context) { + if (log.isTraceEnabled()) { + log.trace("Notify notifyStateChanged " + context); + } ensembleListener.stateChanged(context); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/region/Region.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/region/Region.java index 324a2bfc..b475d2b1 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/region/Region.java +++ b/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 { + /** + * Gets the region and state machine unique id. + * + * @return the region and state machine unique id + */ + String getId(); + /** * Start the region. */ diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java index 33f6a8a8..c66fca69 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java @@ -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 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 extends StateMachineObjectSuppo } @Override - public void addStateChangeInterceptor(StateChangeInterceptor interceptor) { - getStateChangeInterceptors().add(interceptor); + public void addStateMachineInterceptor(StateMachineInterceptor interceptor) { + getStateMachineInterceptors().add(interceptor); + stateMachineExecutor.addStateMachineInterceptor(interceptor); + } + + @Override + public String getId() { + return id; } protected boolean acceptEvent(Message message) { @@ -482,22 +491,29 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo return false; } - private boolean callStateChangeInterceptors(State state, Message message, Transition transition, StateMachine stateMachine) { + private boolean callPreStateChangeInterceptors(State state, Message message, Transition transition, StateMachine 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 state, Message message, Transition transition, StateMachine stateMachine) { + try { + getStateMachineInterceptors().postStateChange(state, message, transition, stateMachine); + } catch (Exception e) { + } + } + private boolean isInitialTransition(Transition transition) { return transition != null && transition.getKind() == TransitionKind.INITIAL; } private void switchToState(State state, Message message, Transition transition, StateMachine 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 extends StateMachineObjectSuppo setCurrentState(state, message, transition, true, stateMachine); } + callPostStateChangeInterceptors(state, message, transition, stateMachine); + stateMachineExecutor.execute(); if (isComplete()) { stop(); diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultExtendedState.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultExtendedState.java index 5d3f1571..51537e2f 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultExtendedState.java +++ b/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 + "]"; + } + } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java index 18ce5f3a..154c3161 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java @@ -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 extends LifecycleObjectSupport im private StateMachineExecutorTransit stateMachineExecutorTransit; + private final StateMachineInterceptorList interceptors = + new StateMachineInterceptorList(); + /** * Instantiates a new default state machine executor. * @@ -154,6 +158,11 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im initialHandled.set(!enabled); } + @Override + public void addStateMachineInterceptor(StateMachineInterceptor interceptor) { + interceptors.add(interceptor); + } + private void handleTriggerTrans(List> trans, Message queuedMessage) { for (Transition t : trans) { if (t == null) { @@ -167,10 +176,17 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im if (!StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) { continue; } + StateContext 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 extends LifecycleObjectSupport im private StateContext buildStateContext(Message message, Transition transition, StateMachine 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()); - return new DefaultStateContext(event, messageHeaders, extendedState, transition, stateMachine); + Map map = new HashMap(messageHeaders); + map.put(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER, stateMachine.getId()); + return new DefaultStateContext(event, new MessageHeaders(map), extendedState, transition, stateMachine); } private void registerTriggerListener() { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java index 25f7214d..0e514b63 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineExecutor.java @@ -86,6 +86,13 @@ public interface StateMachineExecutor { */ void setStateMachineExecutorTransit(StateMachineExecutorTransit stateMachineExecutorTransit); + /** + * Adds the state machine interceptor. + * + * @param interceptor the interceptor + */ + void addStateMachineInterceptor(StateMachineInterceptor interceptor); + /** * Callback interface when executor wants to handle transit. */ diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineInterceptor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineInterceptor.java new file mode 100644 index 00000000..c95cc8cd --- /dev/null +++ b/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 the type of state + * @param the type of event + */ +public interface StateMachineInterceptor { + + /** + * 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 state, Message message, Transition transition, + StateMachine 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 state, Message message, Transition transition, + StateMachine 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 preTransition(StateContext stateContext); + + /** + * Called after of a transition if transition happened. + * + * @param stateContext the state context + * @return the state context + */ + StateContext postTransition(StateContext stateContext); + +} diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateChangeInterceptor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineInterceptorAdapter.java similarity index 50% rename from spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateChangeInterceptor.java rename to spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineInterceptorAdapter.java index 476407b7..15f2a73a 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateChangeInterceptor.java +++ b/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 { +/** + * Adapter helper implementation for {@link StateMachineInterceptor}. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public class StateMachineInterceptorAdapter implements StateMachineInterceptor { - void preStateChange(State state, Message message, Transition transition, StateMachine stateMachine); + @Override + public void preStateChange(State state, Message message, Transition transition, + StateMachine stateMachine) { + } + + @Override + public void postStateChange(State state, Message message, Transition transition, + StateMachine stateMachine) { + } + + @Override + public StateContext preTransition(StateContext stateContext) { + return stateContext; + } + + @Override + public StateContext postTransition(StateContext stateContext) { + return stateContext; + } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineInterceptorList.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineInterceptorList.java new file mode 100644 index 00000000..7aab7b0d --- /dev/null +++ b/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 the type of state + * @param the type of event + */ +public class StateMachineInterceptorList { + + private final List> interceptors = new CopyOnWriteArrayList>(); + + /** + * Sets the interceptors, clears any existing interceptors. + * + * @param interceptors the list of interceptors + * @return true if interceptor list changed as a result of the + * call + */ + public boolean set(List> interceptors) { + synchronized (interceptors) { + interceptors.clear(); + return interceptors.addAll(interceptors); + } + } + + /** + * Adds interceptor to the list. + * + * @param interceptor the interceptor + * @return true (as specified by {@link Collection#add}) + */ + public boolean add(StateMachineInterceptor interceptor) { + return interceptors.add(interceptor); + } + + /** + * Removes interceptor from the list. + * + * @param interceptor the interceptor + * @return true (as specified by {@link Collection#remove}) + */ + public boolean remove(StateMachineInterceptor 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 state, Message message, Transition transition, + StateMachine stateMachine) { + for (StateMachineInterceptor 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 state, Message message, Transition transition, + StateMachine stateMachine) { + for (StateMachineInterceptor interceptor : interceptors) { + interceptor.postStateChange(state, message, transition, stateMachine); + } + } + + /** + * Pre transition. + * + * @param stateContext the state context + * @return the state context + */ + public StateContext preTransition(StateContext stateContext) { + for (StateMachineInterceptor interceptor : interceptors) { + if ((stateContext = interceptor.preTransition(stateContext)) == null) { + break; + } + } + return stateContext; + } + + /** + * Post transition. + * + * @param stateContext the state context + * @return the state context + */ + public StateContext postTransition(StateContext stateContext) { + for (StateMachineInterceptor interceptor : interceptors) { + if ((stateContext = interceptor.postTransition(stateContext)) == null) { + break; + } + } + return stateContext; + } + +} diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineObjectSupport.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineObjectSupport.java index 601e4205..5d1a053d 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineObjectSupport.java +++ b/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 extends LifecycleObjectSup /** Flag for application context events */ private boolean contextEventsEnabled = true; - private final StateChangeInterceptorList interceptors = - new StateChangeInterceptorList(); + private final StateMachineInterceptorList interceptors = + new StateMachineInterceptorList(); /** * Gets the state machine event publisher. @@ -195,11 +193,11 @@ public abstract class StateMachineObjectSupport extends LifecycleObjectSup // re-scheduling is needed. } - protected StateChangeInterceptorList getStateChangeInterceptors() { + protected StateMachineInterceptorList getStateMachineInterceptors() { return interceptors; } - protected void setStateChangeInterceptors(List> interceptors) { + protected void setStateMachineInterceptors(List> interceptors) { Collections.sort(interceptors, new OrderComparator()); this.interceptors.set(interceptors); } @@ -259,53 +257,4 @@ public abstract class StateMachineObjectSupport extends LifecycleObjectSup } - protected class StateChangeInterceptorList { - - private final List> interceptors = new CopyOnWriteArrayList>(); - - /** - * Sets the interceptors, clears any existing interceptors. - * - * @param interceptors the list of interceptors - * @return true if interceptor list changed as a result of the - * call - */ - public boolean set(List> interceptors) { - synchronized (interceptors) { - interceptors.clear(); - return interceptors.addAll(interceptors); - } - } - - /** - * Adds interceptor to the list. - * - * @param interceptor the interceptor - * @return true (as specified by {@link Collection#add}) - */ - public boolean add(StateChangeInterceptor interceptor) { - return interceptors.add(interceptor); - } - - /** - * Removes interceptor from the list. - * - * @param interceptor the interceptor - * @return true (as specified by {@link Collection#remove}) - */ - public boolean remove(StateChangeInterceptor interceptor) { - return interceptors.remove(interceptor); - } - - /** - * Handles the pre state change calls. - */ - void preStateChange(State state, Message message, Transition transition, - StateMachine stateMachine) { - for (StateChangeInterceptor interceptor : interceptors) { - interceptor.preStateChange(state, message, transition, stateMachine); - } - } - - } } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java index c9a3948a..a32797ca 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java +++ b/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 interceptor) { + public void addStateMachineInterceptor(StateMachineInterceptor interceptor) { } @Override @@ -164,6 +164,11 @@ public class StateMachineAccessTests { public void setInitialEnabled(boolean enabled) { } + @Override + public String getId() { + return null; + } + } } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java index 32399129..993fe9fe 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java @@ -48,4 +48,9 @@ public class InMemoryStateMachineEnsemble extends StateMachineEnsembleObje notifyStateChanged(context); } + @Override + public StateMachineContext getState() { + return current; + } + } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateChangeInterceptorTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateChangeInterceptorTests.java index 12510590..7920193c 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateChangeInterceptorTests.java +++ b/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 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 { + private static class TestStateChangeInterceptor implements StateMachineInterceptor { volatile CountDownLatch preStateChangeLatch = new CountDownLatch(1); volatile int preStateChangeCount = 0; @@ -279,6 +280,22 @@ public class StateChangeInterceptorTests extends AbstractStateMachineTests { } + @Override + public void postStateChange(State state, Message message, + Transition transition, StateMachine stateMachine) { + } + + @Override + public StateContext preTransition(StateContext stateContext) { + return stateContext; + } + + @Override + public StateContext postTransition(StateContext stateContext) { + return stateContext; + } + + public void reset(int c1) { preStateChangeLatch = new CountDownLatch(c1); preStateChangeCount = 0; diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateContextExpressionMethodsTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateContextExpressionMethodsTests.java index ae4b86d2..038f1acf 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateContextExpressionMethodsTests.java +++ b/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; + } + } } diff --git a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java index 5fc4a3aa..59ef6575 100644 --- a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java +++ b/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 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 state, Message message, Transition transition, StateMachine stateMachine); + void onPersist(State state, Message message, Transition transition, + StateMachine stateMachine); } - private class PersistingStateChangeInterceptor implements StateChangeInterceptor { + private class PersistingStateChangeInterceptor extends StateMachineInterceptorAdapter { @Override public void preStateChange(State state, Message message, Transition transition, StateMachine stateMachine) { listeners.onPersist(state, message, transition, stateMachine); } - } private class CompositePersistStateChangeListener extends AbstractCompositeListener implements diff --git a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java index 3f252976..a70c3b20 100644 --- a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java +++ b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java @@ -66,6 +66,7 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj private final boolean cleanState; private final StateMachinePersist persist; private final AtomicReference stateRef = new AtomicReference(); + private final AtomicReference notifyRef = new AtomicReference(); private final CuratorWatcher watcher = new StateWatcher(); private PersistentEphemeralNode node; @@ -97,6 +98,7 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj this.memberPath = basePath + "/" + PATH_MEMBERS; this.mutexPath = basePath + "/" + PATH_MUTEX; this.persist = new ZookeeperStateMachinePersist(curatorClient, statePath, logPath, logSize); + setAutoStartup(true); } @Override @@ -106,6 +108,20 @@ public class ZookeeperStateMachineEnsemble 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 extends StateMachineEnsembleObj @Override public void join(StateMachine 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 extends StateMachineEnsembleObj } @Override - public void setState(StateMachineContext context) { + public synchronized void setState(StateMachineContext 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 extends StateMachineEnsembleObj } } + @Override + public StateMachineContext 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 context = persist.read(stat); return new StateWrapper(context, stat.getVersion()); } catch (Exception e) { @@ -175,8 +188,13 @@ public class ZookeeperStateMachineEnsemble 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 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 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 context = ((ZookeeperStateMachinePersist) 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 extends StateMachineEnsembleObj } /** - * Wrapper object for a {@link StateMachineContext}. + * Wrapper object for a {@link StateMachineContext} and its + * current version. */ private class StateWrapper { private final StateMachineContext context; diff --git a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java index 27cf0116..ec303f10 100644 --- a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java +++ b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java @@ -121,9 +121,11 @@ public class ZookeeperStateMachinePersist implements StateMachinePersist read(Stat stat) throws Exception { - byte[] data = curatorClient.getData().storingStatIn(stat).forPath(path); - StateMachineContext context = deserialize(data); - return context; + return deserialize(curatorClient.getData().storingStatIn(stat).forPath(path)); + } + + public StateMachineContext readLog(int version, Stat stat) throws Exception { + return deserialize(curatorClient.getData().storingStatIn(stat).forPath(logPath + "/" + version)); } private byte[] serialize(StateMachineContext context) { diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java index c53f9fb4..ef66f127 100644 --- a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java +++ b/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 ensemble = + new ZookeeperStateMachineEnsemble(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("S1","E1", new HashMap(), 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("S2","E1", new HashMap(), 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("S3","E1", new HashMap(), 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("S4","E1", new HashMap(), 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(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 ensemble = + new ZookeeperStateMachineEnsemble(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("S" + i, "E" + i, + new HashMap(), 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 ensemble = + new ZookeeperStateMachineEnsemble(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("S" + i, "E" + i, + new HashMap(), 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> events = new ArrayList>(); @Override public void stateMachineJoined(StateMachineContext context) { @@ -162,9 +282,16 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { @Override public void stateChanged(StateMachineContext 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 { @@ -230,6 +357,11 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { return null; } + @Override + public String getId() { + return null; + } + } } diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java index 011c3e02..edf64af0 100644 --- a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java +++ b/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 persist = new ZookeeperStateMachinePersist( + 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 contextOut = + new DefaultStateMachineContext("S" + i, "E" + i, new HashMap(), new DefaultExtendedState()); + persist.write(contextOut, stat); + } + + for (int i = 0; i < 10; i++) { + StateMachineContext 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); diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java index 94736065..d42de2fd 100644 --- a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java +++ b/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 config) throws Exception {