diff --git a/build.gradle b/build.gradle index 5a01c1a4..32c59313 100644 --- a/build.gradle +++ b/build.gradle @@ -22,8 +22,16 @@ configure(allprojects) { apply plugin: 'eclipse' apply plugin: 'idea' - sourceCompatibility = 1.6 - targetCompatibility = 1.6 + + compileJava { + sourceCompatibility = 1.7 + targetCompatibility = 1.7 + } + + compileTestJava { + sourceCompatibility = 1.8 + targetCompatibility = 1.8 + } group = 'org.springframework.statemachine' @@ -111,11 +119,29 @@ project('spring-statemachine-core') { } } +project('spring-statemachine-zookeeper') { + description = "Spring State Machine Zookeeper" + + dependencies { + compile project(":spring-statemachine-core") + compile "org.apache.curator:curator-recipes:$curatorVersion" + compile "com.esotericsoftware.kryo:kryo:$kryoVersion" + + testCompile "org.apache.curator:curator-test:$curatorVersion" + testCompile "org.springframework:spring-test:$springVersion" + testCompile "org.hamcrest:hamcrest-core:$hamcrestVersion" + testCompile "org.hamcrest:hamcrest-library:$hamcrestVersion" + testCompile "junit:junit:$junitVersion" + testRuntime("log4j:log4j:$log4jVersion") + } +} + configure(sampleProjects()) { apply plugin: 'spring-boot' configurations.archives.artifacts.removeAll { it.archiveTask.is jar } dependencies { compile project(":spring-statemachine-samples-common") + compile project(":spring-statemachine-zookeeper") testCompile "org.springframework:spring-test:$springVersion" testCompile "org.hamcrest:hamcrest-core:$hamcrestVersion" testCompile "org.hamcrest:hamcrest-library:$hamcrestVersion" diff --git a/gradle.properties b/gradle.properties index 20c25680..5045e16c 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,5 +1,7 @@ version=1.0.0.BUILD-SNAPSHOT springVersion = 4.1.6.RELEASE +curatorVersion = 2.6.0 +kryoVersion = 2.22 hamcrestVersion = 1.3 junitVersion = 4.11 log4jVersion = 1.2.17 diff --git a/settings.gradle b/settings.gradle index db02c02c..ed0bd9ec 100644 --- a/settings.gradle +++ b/settings.gradle @@ -1,6 +1,7 @@ rootProject.name = 'spring-statemachine' include 'spring-statemachine-core' +include 'spring-statemachine-zookeeper' include 'spring-statemachine-samples' include 'spring-statemachine-samples:turnstile' @@ -8,6 +9,7 @@ include 'spring-statemachine-samples:showcase' include 'spring-statemachine-samples:cdplayer' include 'spring-statemachine-samples:tasks' include 'spring-statemachine-samples:washer' +include 'spring-statemachine-samples:zookeeper' rootProject.children.find { if (it.name == 'spring-statemachine-samples') { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachine.java index f45f7b2c..eb39c27c 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachine.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachine.java @@ -15,6 +15,7 @@ */ package org.springframework.statemachine; +import org.springframework.statemachine.access.StateMachineAccessor; import org.springframework.statemachine.region.Region; import org.springframework.statemachine.state.State; @@ -43,4 +44,11 @@ public interface StateMachine extends Region { */ ExtendedState getExtendedState(); + /** + * Gets the state machine accessor. + * + * @return the state machine accessor + */ + StateMachineAccessor getStateMachineAccessor(); + } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineContext.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineContext.java new file mode 100644 index 00000000..7c66d07f --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineContext.java @@ -0,0 +1,65 @@ +/* + * 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; + +import java.util.Map; + +/** + * {@code StateMachineContext} represents a current state of a state machine. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public interface StateMachineContext { + + /** + * Gets the state machine. + * + * @return the state machine + */ + StateMachine getStateMachine(); + + /** + * Gets the state. + * + * @return the state + */ + S getState(); + + /** + * Gets the event. + * + * @return the event + */ + E getEvent(); + + /** + * Gets the event headers. + * + * @return the event headers + */ + Map getEventHeaders(); + + /** + * Gets the extended state. + * + * @return the extended state + */ + ExtendedState getExtendedState(); + +} diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineAccess.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java similarity index 71% rename from spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineAccess.java rename to spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java index e0d943d5..f84c6d35 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineAccess.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java @@ -13,13 +13,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.statemachine.support; +package org.springframework.statemachine.access; +import org.springframework.statemachine.ExtendedState; import org.springframework.statemachine.StateMachine; /** - * Functional interface for {@link StateMachine} to allow more programmetic - * access to underlying functionality. + * Functional interface exposing {@link StateMachine} internals. * * @author Janne Valkealahti * @@ -28,13 +28,6 @@ import org.springframework.statemachine.StateMachine; */ public interface StateMachineAccess { - /** - * Execute given {@link StateMachineFunction} with all recursive regions. - * - * @param stateMachineAccess the state machine access - */ - void doWithAllRegions(StateMachineFunction> stateMachineAccess); - /** * Sets the relay state machine. * @@ -42,4 +35,18 @@ public interface StateMachineAccess { */ void setRelay(StateMachine stateMachine); + /** + * Reset state. + * + * @param state the state + */ + void resetState(S state); + + /** + * Sets the extended state. + * + * @param extendedState the new extended state + */ + void setExtendedState(ExtendedState extendedState); + } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccessor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccessor.java new file mode 100644 index 00000000..25e19b42 --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccessor.java @@ -0,0 +1,49 @@ +/* + * 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.access; + +import java.util.List; + +import org.springframework.statemachine.StateMachine; + +/** + * Functional interface for {@link StateMachine} to allow more programmatic + * access to underlying functionality. Functions prefixed "doWith" will expose + * {@link StateMachineAccess} via {@link StateMachineFunction} for better functional + * access with jdk7. Functions prefixed "with" is better suitable for lambdas. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public interface StateMachineAccessor { + + /** + * Execute given {@link StateMachineFunction} with all recursive regions. + * + * @param stateMachineAccess the state machine access + */ + void doWithAllRegions(StateMachineFunction> stateMachineAccess); + + /** + * Gets all regions. + * + * @return the all regions + */ + List> withAllRegions(); + +} diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineFunction.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineFunction.java similarity index 92% rename from spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineFunction.java rename to spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineFunction.java index eec0e639..ef90c86e 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineFunction.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineFunction.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.statemachine.support; +package org.springframework.statemachine.access; /** * Strategic function interface for applying arbitrary function @@ -22,7 +22,7 @@ package org.springframework.statemachine.support; * @author Janne Valkealahti * * @param the function type - * @see StateMachineAccess + * @see StateMachineAccessor */ public interface StateMachineFunction { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java index f222888b..45d11442 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java @@ -28,6 +28,8 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.messaging.Message; import org.springframework.statemachine.ExtendedState; import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.access.StateMachineAccess; +import org.springframework.statemachine.access.StateMachineFunction; import org.springframework.statemachine.action.Action; import org.springframework.statemachine.config.builders.StateMachineStates; import org.springframework.statemachine.config.builders.StateMachineTransitions; @@ -47,8 +49,6 @@ import org.springframework.statemachine.state.State; import org.springframework.statemachine.state.StateMachineState; import org.springframework.statemachine.support.DefaultExtendedState; import org.springframework.statemachine.support.LifecycleObjectSupport; -import org.springframework.statemachine.support.StateMachineAccess; -import org.springframework.statemachine.support.StateMachineFunction; import org.springframework.statemachine.support.tree.Tree; import org.springframework.statemachine.support.tree.Tree.Node; import org.springframework.statemachine.support.tree.TreeTraverser; @@ -178,15 +178,15 @@ public abstract class AbstractStateMachineFactory extends LifecycleObjectS } // set top-level machine as relay - final StateMachine mm = machine; - ((StateMachineAccess)machine).doWithAllRegions(new StateMachineFunction>() { + final StateMachine fmachine = machine; + fmachine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { @Override - public void apply(StateMachineAccess stateMachineAccess) { - stateMachineAccess.setRelay(mm); + public void apply(StateMachineAccess function) { + function.setRelay(fmachine); } - }); + }); return machine; } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/CompositeEnsembleListener.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/CompositeEnsembleListener.java new file mode 100644 index 00000000..9cc088ae --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/CompositeEnsembleListener.java @@ -0,0 +1,58 @@ +/* + * 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.ensemble; + +import java.util.Iterator; + +import org.springframework.statemachine.StateMachineContext; +import org.springframework.statemachine.listener.AbstractCompositeListener; + +/** + * Default {@link EnsembleListeger} dispatcher. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public class CompositeEnsembleListener extends AbstractCompositeListener> implements + EnsembleListeger { + + @Override + public void stateMachineJoined(StateMachineContext context) { + for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { + EnsembleListeger listener = iterator.next(); + listener.stateMachineJoined(context); + } + } + + @Override + public void stateMachineLeft(StateMachineContext context) { + for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { + EnsembleListeger listener = iterator.next(); + listener.stateMachineLeft(context); + } + } + + @Override + public void stateChanged(StateMachineContext context) { + for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { + EnsembleListeger listener = iterator.next(); + listener.stateChanged(context); + } + } + +} 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 new file mode 100644 index 00000000..854cdccf --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java @@ -0,0 +1,195 @@ +/* + * 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.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.access.StateMachineAccess; +import org.springframework.statemachine.access.StateMachineAccessor; +import org.springframework.statemachine.access.StateMachineFunction; +import org.springframework.statemachine.listener.StateMachineListener; +import org.springframework.statemachine.listener.StateMachineListenerAdapter; +import org.springframework.statemachine.state.State; +import org.springframework.statemachine.support.DefaultStateMachineContext; +import org.springframework.statemachine.support.LifecycleObjectSupport; +import org.springframework.statemachine.transition.Transition; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +/** + * {@code DistributedStateMachine} is wrapping a real {@link StateMachine} and works + * together with a {@link StateMachineEnsemble} order to provide a distributed state + * machine. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +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; + private final LocalStateMachineListener stateMachineListener; + + /** + * Instantiates a new distributed state machine. + * + * @param ensemble the state machine ensemble + * @param delegate the delegating state machine + */ + public DistributedStateMachine(StateMachineEnsemble ensemble, StateMachine delegate) { + Assert.notNull(ensemble, "State machine ensemble must be set"); + Assert.notNull(delegate, "State machine delegate must be set"); + this.ensemble = ensemble; + this.delegate = delegate; + this.listener = new LocalEnsembleListener(); + this.stateMachineListener = new LocalStateMachineListener(); + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + } + + @Override + protected void doStart() { + ensemble.addEnsembleListener(listener); + ensemble.join(this); + delegate.addStateListener(stateMachineListener); + super.doStart(); + } + + @Override + protected void doStop() { + ensemble.removeEnsembleListener(listener); + super.doStop(); + } + + @Override + public boolean sendEvent(Message event) { + return delegate.sendEvent(MessageBuilder.fromMessage(event).setHeader("uuid", uuid).build()); + } + + @Override + public boolean sendEvent(E event) { + return sendEvent(MessageBuilder.withPayload(event).build()); + } + + @Override + public State getState() { + return delegate.getState(); + } + + @Override + public Collection> getStates() { + return delegate.getStates(); + } + + @Override + public Collection> getTransitions() { + return delegate.getTransitions(); + } + + @Override + public boolean isComplete() { + return delegate.isComplete(); + } + + @Override + public void addStateListener(StateMachineListener listener) { + delegate.addStateListener(listener); + } + + @Override + public void removeStateListener(StateMachineListener listener) { + delegate.removeStateListener(listener); + } + + @Override + public State getInitialState() { + return delegate.getInitialState(); + } + + @Override + public ExtendedState getExtendedState() { + return delegate.getExtendedState(); + } + + @Override + public StateMachineAccessor getStateMachineAccessor() { + return delegate.getStateMachineAccessor(); + } + + private class LocalStateMachineListener extends StateMachineListenerAdapter { + + @Override + public void stateChanged(StateContext context) { + if (ObjectUtils.nullSafeEquals(uuid, context.getMessageHeader("uuid"))) { + ensemble.setState(new DefaultStateMachineContext(delegate, context.getTransition().getTarget() + .getId(), context.getEvent(), context.getMessageHeaders(), context.getExtendedState())); + } + } + } + + private class LocalEnsembleListener implements EnsembleListeger { + + @Override + public void stateMachineJoined(final StateMachineContext context) { + if (context != null) { + // I'm now successfully joined, so set delegating + // sm to current known state by a context. + + delegate.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { + + @Override + public void apply(StateMachineAccess function) { + function.resetState(context.getState()); + function.setExtendedState(context.getExtendedState()); + } + + }); + } + log.info("Requesting to start delegating state machine " + delegate); + delegate.start(); + } + + @Override + public void stateMachineLeft(StateMachineContext context) { + log.info("Requesting to stop delegating state machine " + delegate); + delegate.stop(); + } + + @Override + public void stateChanged(StateMachineContext context) { + delegate.sendEvent(MessageBuilder.withPayload(context.getEvent()).copyHeaders(context.getEventHeaders()).build()); + } + + } + +} 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 new file mode 100644 index 00000000..00c5c8e5 --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java @@ -0,0 +1,36 @@ +/* + * 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.ensemble; + +import org.springframework.statemachine.StateMachineContext; + +/** + * {@code EnsembleListeger} for various ensemble events. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public interface EnsembleListeger { + + void stateMachineJoined(StateMachineContext context); + + void stateMachineLeft(StateMachineContext 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 new file mode 100644 index 00000000..fbaf263f --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java @@ -0,0 +1,67 @@ +/* + * 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.ensemble; + +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineContext; + +/** + * {@code StateMachineEnsemble} is a contract between a {@link StateMachine} and + * arbitrary ensemble of other {@link StateMachine}s. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public interface StateMachineEnsemble { + + /** + * Request a join to a state machine ensemble. + * + * @param stateMachine the state machine + */ + void join(StateMachine stateMachine); + + /** + * Request a leave from an ensemble. + * + * @param stateMachine the state machine + */ + void leave(StateMachine stateMachine); + + /** + * Adds the ensemble listener. + * + * @param listener the listener + */ + void addEnsembleListener(EnsembleListeger listener); + + /** + * Removes the ensemble listener. + * + * @param listener the listener + */ + void removeEnsembleListener(EnsembleListeger listener); + + /** + * Sets the state. + * + * @param context the context + */ + void setState(StateMachineContext context); + +} 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 new file mode 100644 index 00000000..dec8cac4 --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java @@ -0,0 +1,62 @@ +/* + * 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.ensemble; + +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineContext; +import org.springframework.statemachine.support.LifecycleObjectSupport; + +/** + * Support class for implementing {@link StateMachineEnsemble}s. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public abstract class StateMachineEnsembleObjectSupport extends LifecycleObjectSupport implements StateMachineEnsemble { + + private final CompositeEnsembleListener ensembleListener = new CompositeEnsembleListener(); + + @Override + public abstract void join(StateMachine stateMachine); + + @Override + public abstract void leave(StateMachine stateMachine); + + @Override + public void addEnsembleListener(EnsembleListeger listener) { + ensembleListener.register(listener); + } + + @Override + public void removeEnsembleListener(EnsembleListeger listener) { + ensembleListener.unregister(listener); + } + + protected void notifyJoined(StateMachineContext context) { + ensembleListener.stateMachineJoined(context); + } + + protected void notifyLeft(StateMachineContext context) { + ensembleListener.stateMachineLeft(context); + } + + protected void notifyStateChanged(StateMachineContext context) { + ensembleListener.stateChanged(context); + } + +} diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachinePersist.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachinePersist.java new file mode 100644 index 00000000..e7e5f793 --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachinePersist.java @@ -0,0 +1,47 @@ +/* + * 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.ensemble; + +import org.springframework.statemachine.StateMachineContext; + +/** + * {@code StateMachinePersist} is an interface handling serialization + * logic of a {@link StateMachineContext}. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public interface StateMachinePersist { + + /** + * Serialize a {@link StateMachineContext}. + * + * @param context the state machine context + * @return the serialized data + */ + byte[] serialize(StateMachineContext context); + + /** + * Deserialize a data into a {@link StateMachineContext}. + * + * @param data the data + * @return the state machine context + */ + StateMachineContext deserialize(byte[] data); + +} diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/CompositeStateMachineListener.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/CompositeStateMachineListener.java index b6e37bce..f937c297 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/CompositeStateMachineListener.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/CompositeStateMachineListener.java @@ -17,6 +17,7 @@ package org.springframework.statemachine.listener; import java.util.Iterator; +import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; @@ -40,6 +41,14 @@ public class CompositeStateMachineListener extends AbstractCompositeListene } } + @Override + public void stateChanged(StateContext context) { + for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { + StateMachineListener listener = iterator.next(); + listener.stateChanged(context); + } + } + @Override public void stateEntered(State state) { for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListener.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListener.java index 74038adf..e95a3559 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListener.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListener.java @@ -15,6 +15,7 @@ */ package org.springframework.statemachine.listener; +import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; @@ -37,6 +38,13 @@ public interface StateMachineListener { */ void stateChanged(State from, State to); + /** + * Notified when state is changed. + * + * @param context the state context + */ + void stateChanged(StateContext context); + /** * Notified when state is entered. * diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListenerAdapter.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListenerAdapter.java index ba49ba3b..ce73e248 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListenerAdapter.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListenerAdapter.java @@ -15,6 +15,7 @@ */ package org.springframework.statemachine.listener; +import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; @@ -34,6 +35,10 @@ public class StateMachineListenerAdapter implements StateMachineListener from, State to) { } + @Override + public void stateChanged(StateContext context) { + } + @Override public void stateEntered(State state) { } 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 6c57d328..cc9222b0 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 @@ -37,6 +37,9 @@ 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.access.StateMachineAccess; +import org.springframework.statemachine.access.StateMachineAccessor; +import org.springframework.statemachine.access.StateMachineFunction; import org.springframework.statemachine.annotation.OnTransition; import org.springframework.statemachine.listener.StateMachineListener; import org.springframework.statemachine.processor.StateMachineHandler; @@ -83,7 +86,7 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo private final Message initialEvent; - private final ExtendedState extendedState; + private ExtendedState extendedState; private volatile State currentState; @@ -165,6 +168,11 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo return extendedState; } + @Override + public void setExtendedState(ExtendedState extendedState) { + this.extendedState = extendedState; + } + public void setHistoryState(PseudoState history) { this.history = history; } @@ -304,21 +312,46 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo @SuppressWarnings("unchecked") @Override - public void doWithAllRegions(StateMachineFunction> stateMachineAccess) { - stateMachineAccess.apply(this); - for (State state : states) { - if (state.isSubmachineState()) { - StateMachine submachine = ((AbstractState)state).getSubmachine(); - if (submachine instanceof StateMachineAccess) { - ((StateMachineAccess)submachine).doWithAllRegions(stateMachineAccess); - } - } else if (state.isOrthogonal()) { - Collection> regions = ((AbstractState)state).getRegions(); - for (Region region : regions) { - ((StateMachineAccess)region).doWithAllRegions(stateMachineAccess); + public StateMachineAccessor getStateMachineAccessor() { + // TODO: needs cleaning and perhaps not an anonymous function + return new StateMachineAccessor() { + + @Override + public void doWithAllRegions(StateMachineFunction> stateMachineAccess) { + stateMachineAccess.apply(AbstractStateMachine.this); + for (State state : states) { + if (state.isSubmachineState()) { + StateMachine submachine = ((AbstractState) state).getSubmachine(); + submachine.getStateMachineAccessor().doWithAllRegions(stateMachineAccess); + } else if (state.isOrthogonal()) { + Collection> regions = ((AbstractState) state).getRegions(); + for (Region region : regions) { + ((StateMachine)region).getStateMachineAccessor().doWithAllRegions(stateMachineAccess); + } + } } } - } + + @Override + public List> withAllRegions() { + List> list = new ArrayList>(); + list.add(AbstractStateMachine.this); + for (State state : states) { + if (state.isSubmachineState()) { + StateMachine submachine = ((AbstractState) state).getSubmachine(); + if (submachine instanceof StateMachineAccess) { + list.add((StateMachineAccess)submachine); + } + } else if (state.isOrthogonal()) { + Collection> regions = ((AbstractState) state).getRegions(); + for (Region region : regions) { + list.add((StateMachineAccess) region); + } + } + } + return list; + } + }; } @Override @@ -353,6 +386,16 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo return buf.toString(); } + @Override + public void resetState(S state) { + for (State s : getStates()) { + if (s.getId().equals(state)) { + currentState = s; + break; + } + } + } + protected boolean acceptEvent(Message message) { boolean accepted = currentState.sendEvent(message); @@ -476,6 +519,8 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo } entryToState(state, message, transition, stateMachine); notifyStateChanged(notifyFrom, state); + StateContext stateContext = buildStateContext(message, transition, stateMachine); + notifyStateChanged(stateContext); } else if (currentState != null) { if (findDeep != null) { if (exit) { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineContext.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineContext.java new file mode 100644 index 00000000..c66a2311 --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineContext.java @@ -0,0 +1,82 @@ +/* + * 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.Map; + +import org.springframework.statemachine.ExtendedState; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineContext; + +/** + * Default implementation of a {@link StateMachineContext}. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public class DefaultStateMachineContext implements StateMachineContext { + + private final StateMachine stateMachine; + private final S state; + private final E event; + private final Map eventHeaders; + private final ExtendedState extendedState; + + /** + * Instantiates a new default state machine context. + * + * @param stateMachine the state machine + * @param state the state + * @param event the event + * @param eventHeaders the event headers + * @param extendedState the extended state + */ + public DefaultStateMachineContext(StateMachine stateMachine, S state, E event, Map eventHeaders, ExtendedState extendedState) { + this.stateMachine = stateMachine; + this.state = state; + this.event = event; + this.eventHeaders = eventHeaders; + this.extendedState = extendedState; + } + + @Override + public StateMachine getStateMachine() { + return stateMachine; + } + + @Override + public S getState() { + return state; + } + + @Override + public E getEvent() { + return event; + } + + @Override + public Map getEventHeaders() { + return eventHeaders; + } + + @Override + public ExtendedState getExtendedState() { + return extendedState; + } + +} 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 4be50abd..6a3cd680 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 @@ -17,6 +17,7 @@ package org.springframework.statemachine.support; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.event.StateMachineEventPublisher; import org.springframework.statemachine.listener.CompositeStateMachineListener; @@ -95,6 +96,16 @@ public abstract class StateMachineObjectSupport extends LifecycleObjectSup } } + protected void notifyStateChanged(StateContext context) { + stateListener.stateChanged(context); +// if (contextEventsEnabled) { +// StateMachineEventPublisher eventPublisher = getStateMachineEventPublisher(); +// if (eventPublisher != null) { +// eventPublisher.publishStateChanged(this, source, target); +// } +// } + } + protected void notifyStateEntered(State state) { stateListener.stateEntered(state); if (contextEventsEnabled) { @@ -188,6 +199,11 @@ public abstract class StateMachineObjectSupport extends LifecycleObjectSup stateChangedInRelay(); } + @Override + public void stateChanged(StateContext context) { + stateListener.stateChanged(context); + } + @Override public void stateEntered(State state) { stateListener.stateEntered(state); 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 new file mode 100644 index 00000000..591e352b --- /dev/null +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java @@ -0,0 +1,154 @@ +/* + * 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.access; + +import static org.hamcrest.Matchers.sameInstance; +import static org.junit.Assert.assertThat; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.junit.Test; +import org.springframework.messaging.Message; +import org.springframework.statemachine.ExtendedState; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.listener.StateMachineListener; +import org.springframework.statemachine.state.State; +import org.springframework.statemachine.transition.Transition; + +public class StateMachineAccessTests { + + @Test + public void testDoWithAllRegionsSetRelay() { + MockStateMachine mock = new MockStateMachine(); + final StateMachine stateMachine = mock; + stateMachine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { + + @Override + public void apply(StateMachineAccess function) { + function.setRelay(stateMachine); + } + + }); + + assertThat(mock.relay, sameInstance(stateMachine)); + } + + @Test + public void testGetAllRegionsSetRelay() { + MockStateMachine mock = new MockStateMachine(); + final StateMachine stateMachine = mock; + stateMachine.getStateMachineAccessor().withAllRegions().stream() + .forEach(access -> access.setRelay(stateMachine)); + + assertThat(mock.relay, sameInstance(stateMachine)); + } + + private static class MockStateMachine implements StateMachine, StateMachineAccess { + + StateMachine relay; + + @Override + public StateMachineAccessor getStateMachineAccessor() { + return new StateMachineAccessor() { + + @Override + public void doWithAllRegions(StateMachineFunction> stateMachineAccess) { + stateMachineAccess.apply(MockStateMachine.this); + } + + @Override + public List> withAllRegions() { + List> list = new ArrayList>(); + list.add(MockStateMachine.this); + return list; + } + }; + } + + @Override + public void setRelay(StateMachine stateMachine) { + this.relay = stateMachine; + } + + @Override + public void resetState(String state) { + } + + @Override + public void setExtendedState(ExtendedState extendedState) { + } + + @Override + public void start() { + } + + @Override + public void stop() { + } + + @Override + public boolean sendEvent(Message event) { + return false; + } + + @Override + public boolean sendEvent(String event) { + return false; + } + + @Override + public State getState() { + return null; + } + + @Override + public Collection> getStates() { + return null; + } + + @Override + public Collection> getTransitions() { + return null; + } + + @Override + public boolean isComplete() { + return false; + } + + @Override + public void addStateListener(StateMachineListener listener) { + } + + @Override + public void removeStateListener(StateMachineListener listener) { + } + + @Override + public State getInitialState() { + return null; + } + + @Override + public ExtendedState getExtendedState() { + return null; + } + + } + +} diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/DistributedStateMachineTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/DistributedStateMachineTests.java new file mode 100644 index 00000000..a49bbf97 --- /dev/null +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/DistributedStateMachineTests.java @@ -0,0 +1,136 @@ +/* + * 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.ensemble; + +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.junit.Assert.assertThat; + +import org.junit.Test; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.statemachine.AbstractStateMachineTests; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.config.EnableStateMachine; +import org.springframework.statemachine.config.StateMachineConfigurerAdapter; +import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; +import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; + +public class DistributedStateMachineTests extends AbstractStateMachineTests { + + @Override + protected AnnotationConfigApplicationContext buildContext() { + return new AnnotationConfigApplicationContext(); + } + + @Test + @SuppressWarnings("unchecked") + public void testMachines() { + context.register(Config1.class, Config2.class); + context.refresh(); + StateMachine machine1 = + context.getBean("sm1", StateMachine.class); + StateMachine machine2 = + context.getBean("sm2", StateMachine.class); + + StateMachineEnsemble ensemble = new InMemoryStateMachineEnsemble(); + + DistributedStateMachine machine1s = + new DistributedStateMachine(ensemble, machine1); + + DistributedStateMachine machine2s = + new DistributedStateMachine(ensemble, machine2); + + machine1s.afterPropertiesSet(); + machine2s.afterPropertiesSet(); + + machine1s.start(); + machine2s.start(); + + machine1s.sendEvent("E1"); + assertThat(machine1.getState().getIds(), containsInAnyOrder("S1")); + assertThat(machine2.getState().getIds(), containsInAnyOrder("S1")); + + machine2s.sendEvent("E2"); + assertThat(machine1.getState().getIds(), containsInAnyOrder("S2")); + assertThat(machine2.getState().getIds(), containsInAnyOrder("S2")); + } + + @Test + @SuppressWarnings("unchecked") + public void testJoin() { + context.register(Config1.class, Config2.class); + context.refresh(); + StateMachine machine1 = + context.getBean("sm1", StateMachine.class); + StateMachine machine2 = + context.getBean("sm2", StateMachine.class); + + StateMachineEnsemble ensemble = new InMemoryStateMachineEnsemble(); + + DistributedStateMachine machine1s = + new DistributedStateMachine(ensemble, machine1); + machine1s.afterPropertiesSet(); + machine1s.start(); + + machine1s.sendEvent("E1"); + assertThat(machine1.getState().getIds(), containsInAnyOrder("S1")); + + DistributedStateMachine machine2s = + new DistributedStateMachine(ensemble, machine2); + machine2s.afterPropertiesSet(); + machine2s.start(); + + assertThat(machine2.getState().getIds(), containsInAnyOrder("S1")); + } + + @Configuration + @EnableStateMachine(name = "sm1") + static class Config1 extends SharedConfig { + } + + @Configuration + @EnableStateMachine(name = "sm2") + static class Config2 extends SharedConfig { + } + + static class SharedConfig extends StateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial("SI") + .state("S1") + .state("S2"); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions + .withExternal() + .source("SI") + .target("S1") + .event("E1") + .and() + .withExternal() + .source("S1") + .target("S2") + .event("E2"); + } + + } + +} 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 new file mode 100644 index 00000000..32399129 --- /dev/null +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java @@ -0,0 +1,51 @@ +/* + * 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.ensemble; + +import java.util.HashSet; +import java.util.Set; + +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineContext; + +public class InMemoryStateMachineEnsemble extends StateMachineEnsembleObjectSupport { + + private final Set> joined = new HashSet>(); + + private StateMachineContext current; + + @Override + public void join(StateMachine stateMachine) { + if (!joined.contains(stateMachine)) { + joined.add(stateMachine); + notifyJoined(current); + } + } + + @Override + public void leave(StateMachine stateMachine) { + if (joined.remove(stateMachine)) { + notifyLeft(current); + } + } + + @Override + public void setState(StateMachineContext context) { + current = context; + notifyStateChanged(context); + } + +} diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/listener/ListenerTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/listener/ListenerTests.java index dcbcc752..146cce2f 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/listener/ListenerTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/listener/ListenerTests.java @@ -131,6 +131,10 @@ public class ListenerTests extends AbstractStateMachineTests { states.add(new Holder(from, to)); } + @Override + public void stateChanged(StateContext context) { + } + @Override public void stateEntered(State state) { } 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 37eb81f3..ae4b86d2 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 @@ -33,6 +33,7 @@ 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.access.StateMachineAccessor; import org.springframework.statemachine.action.Action; import org.springframework.statemachine.listener.StateMachineListener; import org.springframework.statemachine.state.EnumState; @@ -131,6 +132,11 @@ public class StateContextExpressionMethodsTests { ArrayList> events = new ArrayList>(); + @Override + public StateMachineAccessor getStateMachineAccessor() { + return null; + } + @Override public void start() { } diff --git a/spring-statemachine-samples/build.gradle b/spring-statemachine-samples/build.gradle index 17be701b..14b92a4d 100644 --- a/spring-statemachine-samples/build.gradle +++ b/spring-statemachine-samples/build.gradle @@ -19,3 +19,7 @@ project('spring-statemachine-samples-tasks') { project('spring-statemachine-samples-washer') { description = 'Spring State Machine History State Sample' } + +project('spring-statemachine-samples-zookeeper') { + description = 'Spring State Machine Distributed Sample' +} diff --git a/spring-statemachine-samples/zookeeper/.gitignore b/spring-statemachine-samples/zookeeper/.gitignore new file mode 100644 index 00000000..70e6e4b8 --- /dev/null +++ b/spring-statemachine-samples/zookeeper/.gitignore @@ -0,0 +1,19 @@ +.gradle +bin +build +.settings +.classpath +.springBeans +.project +*.iml +*.ipr +*.iws +metastore_db +/samples/pig-scripting/src/main/resources/ml-100k.zip +/samples/pig-scripting/src/main/resources/ml-100k/u.data +/src/test/resources/s3.properties +/.idea/ +.DS_Store +/out/ +target +*.log diff --git a/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/Application.java b/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/Application.java new file mode 100644 index 00000000..5b11be43 --- /dev/null +++ b/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/Application.java @@ -0,0 +1,121 @@ +/* + * 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 demo.zookeeper; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.shell.Bootstrap; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.config.EnableStateMachine; +import org.springframework.statemachine.config.StateMachineConfigurerAdapter; +import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; +import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; +import org.springframework.statemachine.ensemble.DistributedStateMachine; +import org.springframework.statemachine.zookeeper.ZookeeperStateMachineEnsemble; + +@Configuration +public class Application { + + @Configuration + static class ZkConfig { + + @Qualifier("internalStateMachine") + @Autowired + StateMachine internalMachine; + + @Bean + public StateMachine stateMachine() throws Exception { + DistributedStateMachine machine = + new DistributedStateMachine(ensemble(), internalMachine); + return machine; + } + + @Bean + public ZookeeperStateMachineEnsemble ensemble() throws Exception { + ZookeeperStateMachineEnsemble ensemble = + new ZookeeperStateMachineEnsemble(curatorClient(), "/foo"); + return ensemble; + } + + @Bean(destroyMethod = "close") + public CuratorFramework curatorClient() throws Exception { + CuratorFramework client = CuratorFrameworkFactory.builder().defaultData(new byte[0]) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) + .connectString("localhost:2181").build(); + // for testing we start it here, thought initiator + // is trying to start it if not already done + client.start(); + return client; + } + + } + + +//tag::snippetA[] + @Configuration + @EnableStateMachine(name="internalStateMachine") + static class StateMachineConfig + extends StateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) + throws Exception { + states + .withStates() + .initial("LOCKED") + .state("UNLOCKED"); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) + throws Exception { + transitions + .withExternal() + .source("LOCKED") + .target("UNLOCKED") + .event("COIN") + .and() + .withExternal() + .source("UNLOCKED") + .target("LOCKED") + .event("PUSH"); + } + + } +//end::snippetA[] + +//tag::snippetB[] + public static enum States { + LOCKED, UNLOCKED + } +//end::snippetB[] + +//tag::snippetC[] + public static enum Events { + COIN, PUSH + } +//end::snippetC[] + + public static void main(String[] args) throws Exception { + Bootstrap.main(args); + } + +} diff --git a/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/StateMachineCommands.java b/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/StateMachineCommands.java new file mode 100644 index 00000000..bbb4e276 --- /dev/null +++ b/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/StateMachineCommands.java @@ -0,0 +1,33 @@ +/* + * 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 demo.zookeeper; + +import org.springframework.shell.core.annotation.CliCommand; +import org.springframework.shell.core.annotation.CliOption; +import org.springframework.stereotype.Component; + +import demo.AbstractStateMachineCommands; + +@Component +public class StateMachineCommands extends AbstractStateMachineCommands { + + @CliCommand(value = "sm event", help = "Sends an event to a state machine") + public String event(@CliOption(key = { "", "event" }, mandatory = true, help = "The event") final String event) { + getStateMachine().sendEvent(event); + return "Event " + event + " send"; + } + +} \ No newline at end of file diff --git a/spring-statemachine-samples/zookeeper/src/main/resources/META-INF/spring/spring-shell-plugin.xml b/spring-statemachine-samples/zookeeper/src/main/resources/META-INF/spring/spring-shell-plugin.xml new file mode 100644 index 00000000..1fc09f1f --- /dev/null +++ b/spring-statemachine-samples/zookeeper/src/main/resources/META-INF/spring/spring-shell-plugin.xml @@ -0,0 +1,8 @@ + + + + \ No newline at end of file diff --git a/spring-statemachine-samples/zookeeper/src/main/resources/statechartmodel.txt b/spring-statemachine-samples/zookeeper/src/main/resources/statechartmodel.txt new file mode 100644 index 00000000..a551ef9a --- /dev/null +++ b/spring-statemachine-samples/zookeeper/src/main/resources/statechartmodel.txt @@ -0,0 +1,19 @@ ++----------------------------------------------------------------+ +| SM | ++----------------------------------------------------------------+ +| | +| +----------------+ +----------------+ | +| *-->| LOCKED | | UNLOCKED | | +| +----------------+ +----------------+ | +| +---| entry/ | | entry/ |---+ | +| | | exit/ | | exit/ | | | +| | | | | | | | +| PUSH| | |---COIN-->| | |COIN | +| | | | | | | | +| | | | | | | | +| | | |<--PUSH---| | | | +| +-->| | | |<--+ | +| | | | | | +| +----------------+ +----------------+ | +| | ++----------------------------------------------------------------+ diff --git a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersist.java b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersist.java new file mode 100644 index 00000000..6d028dc3 --- /dev/null +++ b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersist.java @@ -0,0 +1,92 @@ +/* + * 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.zookeeper; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; + +import org.springframework.statemachine.StateMachineContext; +import org.springframework.statemachine.ensemble.StateMachinePersist; +import org.springframework.statemachine.support.DefaultStateMachineContext; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Serializer; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; + +/** + * {@link StateMachinePersist} using kroy libraries as a backing + * serialization technique. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public class KryoStateMachinePersist implements StateMachinePersist { + + private static final ThreadLocal kryoThreadLocal = new ThreadLocal() { + + @SuppressWarnings("rawtypes") + @Override + protected Kryo initialValue() { + Kryo kryo = new Kryo(); + kryo.addDefaultSerializer(StateMachineContext.class, new StateMachineContextSerializer()); + return kryo; + } + }; + + @Override + public byte[] serialize(StateMachineContext context) { + Kryo kryo = kryoThreadLocal.get(); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + Output output = new Output(out); + kryo.writeObject(output, context); + output.close(); + return out.toByteArray(); + } + + @SuppressWarnings("unchecked") + @Override + public StateMachineContext deserialize(byte[] data) { + if (data == null || data.length == 0) { + return null; + } + Kryo kryo = kryoThreadLocal.get(); + ByteArrayInputStream in = new ByteArrayInputStream(data); + Input input = new Input(in); + return kryo.readObject(input, StateMachineContext.class); + } + + private static class StateMachineContextSerializer extends Serializer> { + + @Override + public void write(Kryo kryo, Output output, StateMachineContext context) { + kryo.writeClassAndObject(output, context.getEvent()); + kryo.writeClassAndObject(output, context.getState()); + } + + @SuppressWarnings("unchecked") + @Override + public StateMachineContext read(Kryo kryo, Input input, Class> clazz) { + E event = (E) kryo.readClassAndObject(input); + S state = (S) kryo.readClassAndObject(input); + return new DefaultStateMachineContext(null, state, event, null, null); + } + + } + +} 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 new file mode 100644 index 00000000..32282f29 --- /dev/null +++ b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java @@ -0,0 +1,188 @@ +/* + * 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.zookeeper; + +import java.util.Collection; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.api.CuratorWatcher; +import org.apache.curator.framework.api.transaction.CuratorTransaction; +import org.apache.curator.framework.api.transaction.CuratorTransactionResult; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.data.Stat; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineContext; +import org.springframework.statemachine.StateMachineException; +import org.springframework.statemachine.ensemble.StateMachineEnsemble; +import org.springframework.statemachine.ensemble.StateMachineEnsembleObjectSupport; +import org.springframework.statemachine.ensemble.StateMachinePersist; + +/** + * {@link StateMachineEnsemble} backed by a zookeeper. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObjectSupport { + + private final static Log log = LogFactory.getLog(ZookeeperStateMachineEnsemble.class); + private final CuratorFramework curatorClient; + private final String basePath; + private final String statePath; + private final String logPath; + private final StateMachinePersist persist = new KryoStateMachinePersist(); + private final AtomicReference stateRef = new AtomicReference(); + private final CuratorWatcher watcher = new StateWatcher(); + + /** + * Instantiates a new zookeeper state machine ensemble. + * + * @param curatorClient the curator client + * @param basePath the base zookeeper path + */ + public ZookeeperStateMachineEnsemble(CuratorFramework curatorClient, String basePath) { + this.curatorClient = curatorClient; + this.basePath = basePath; + this.statePath = basePath + "/current"; + this.logPath = basePath + "/log"; + } + + @Override + protected void onInit() throws Exception { + initPaths(); + } + + @Override + protected void doStart() { + } + + @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); + } + + @Override + public void leave(StateMachine stateMachine) { + } + + @Override + public void setState(StateMachineContext context) { + byte[] data = persist.serialize(context); + CuratorTransaction tx = curatorClient.inTransaction(); + try { + Collection results = tx.setData().forPath(statePath, data).and().commit(); + int version = results.iterator().next().getResultStat().getVersion(); + stateRef.set(new StateWrapper(context, version)); + } catch (Exception e) { + throw new StateMachineException("Error persisting data", e); + } + } + + private StateWrapper readCurrentContext() { + try { + Stat stat = new Stat(); + byte[] data = curatorClient.getData().storingStatIn(stat).usingWatcher(watcher).forPath(statePath); + StateMachineContext context = persist.deserialize(data); + return new StateWrapper(context, stat.getVersion()); + } catch (Exception e) { + throw new StateMachineException("Error reading data", e); + } + } + + private void initPaths() { + try { + if (curatorClient.checkExists().forPath(statePath) == null) { + curatorClient.inTransaction() + .create().forPath(basePath) + .and() + .create().forPath(statePath) + .and() + .create().forPath(logPath) + .and() + .commit(); + } + } catch (KeeperException.NodeExistsException e) { + // ignore, already created + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + private class StateWatcher implements CuratorWatcher { + + @Override + public void process(WatchedEvent event) throws Exception { + if (log.isTraceEnabled()) { + log.trace("Process WatchedEvent: " + event); + } + switch (event.getType()) { + case NodeDataChanged: + StateWrapper currentStateWrapper = stateRef.get(); + StateWrapper newStateWrapper = readCurrentContext(); + if (log.isTraceEnabled()) { + log.trace("NodeDataChanged currentStateWrapper=" + currentStateWrapper + " newStateWrapper=" + newStateWrapper); + } + if (currentStateWrapper.version + 1 == newStateWrapper.version + && stateRef.compareAndSet(currentStateWrapper, newStateWrapper)) { + if (log.isTraceEnabled()) { + log.trace("Notify state change with new context"); + } + notifyStateChanged(newStateWrapper.context); + } + break; + default: + curatorClient.checkExists().usingWatcher(this).forPath(statePath); + break; + } + } + + } + + /** + * Wrapper object for a {@link StateMachineContext}. + */ + private class StateWrapper { + private final StateMachineContext context; + private final int version; + + public StateWrapper(StateMachineContext context, int version) { + this.context = context; + this.version = version; + } + + @Override + public String toString() { + return "StateWrapper [context=" + context + ", version=" + version + "]"; + } + } + +} diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/AbstractZookeeperTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/AbstractZookeeperTests.java new file mode 100644 index 00000000..2848b05e --- /dev/null +++ b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/AbstractZookeeperTests.java @@ -0,0 +1,104 @@ +/* + * 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.zookeeper; + +import java.io.IOException; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.apache.curator.test.TestingServer; +import org.junit.After; +import org.junit.Before; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +public abstract class AbstractZookeeperTests { + + protected AnnotationConfigApplicationContext context; + + @Before + public void setup() { + context = buildContext(); + } + + @After + public void clean() { + if (context != null) { + context.close(); + } + } + + protected AnnotationConfigApplicationContext buildContext() { + return null; + } + + @Configuration + protected static class ZkServerConfig { + + @Bean + public TestingServerWrapper testingServerWrapper() throws Exception { + return new TestingServerWrapper(); + } + + } + + @Configuration + protected static class BaseConfig { + + @Autowired + TestingServerWrapper testingServerWrapper; + + @Bean(destroyMethod = "close") + public CuratorFramework curatorClient() throws Exception { + CuratorFramework client = CuratorFrameworkFactory.builder().defaultData(new byte[0]) + .retryPolicy(new ExponentialBackoffRetry(1000, 3)) + .connectString("localhost:" + testingServerWrapper.getPort()).build(); + // for testing we start it here, thought initiator + // is trying to start it if not already done + client.start(); + return client; + } + + } + + protected static class TestingServerWrapper implements DisposableBean { + + TestingServer testingServer; + + public TestingServerWrapper() throws Exception { + this.testingServer = new TestingServer(true); + } + + @Override + public void destroy() throws Exception { + try { + testingServer.close(); + } + catch (IOException e) { + } + } + + public int getPort() { + return testingServer.getPort(); + } + + } + +} diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersistTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersistTests.java new file mode 100644 index 00000000..a52662a4 --- /dev/null +++ b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersistTests.java @@ -0,0 +1,40 @@ +/* + * 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.zookeeper; + +import static org.hamcrest.Matchers.is; +import static org.junit.Assert.assertThat; + +import org.junit.Test; +import org.springframework.statemachine.StateMachineContext; +import org.springframework.statemachine.ensemble.StateMachinePersist; +import org.springframework.statemachine.support.DefaultStateMachineContext; + +public class KryoStateMachinePersistTests { + + @Test + public void testStateEvent() { + StateMachinePersist persist = new KryoStateMachinePersist(); + StateMachineContext contextOut = + new DefaultStateMachineContext(null, "S1", "E1", null, null); + byte[] data = persist.serialize(contextOut); + StateMachineContext contextIn = persist.deserialize(data); + + assertThat(contextOut.getState(), is(contextIn.getState())); + assertThat(contextOut.getEvent(), is(contextIn.getEvent())); + } + +} 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 new file mode 100644 index 00000000..c6a3d1fa --- /dev/null +++ b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java @@ -0,0 +1,212 @@ +/* + * 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.zookeeper; + +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.notNullValue; +import static org.junit.Assert.assertThat; + +import java.util.Collection; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.curator.framework.CuratorFramework; +import org.junit.Test; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.messaging.Message; +import org.springframework.statemachine.ExtendedState; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineContext; +import org.springframework.statemachine.access.StateMachineAccessor; +import org.springframework.statemachine.ensemble.EnsembleListeger; +import org.springframework.statemachine.listener.StateMachineListener; +import org.springframework.statemachine.state.State; +import org.springframework.statemachine.support.DefaultStateMachineContext; +import org.springframework.statemachine.transition.Transition; + +public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { + + @Test + public void testInitStart() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + + ensemble.afterPropertiesSet(); + + assertThat(curatorClient.checkExists().forPath("/foo/current"), notNullValue()); + assertThat(curatorClient.checkExists().forPath("/foo/log"), notNullValue()); + + ensemble.start(); + } + + @Test + public void testPersist() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + + ensemble.afterPropertiesSet(); + + assertThat(curatorClient.checkExists().forPath("/foo/current"), notNullValue()); + + ensemble.setState(new DefaultStateMachineContext(null, "S1","E1", null, null)); + ensemble.setState(new DefaultStateMachineContext(null, "S2","E1", null, null)); + + } + + @Test + public void testReceiveEvents() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ZookeeperStateMachineEnsemble ensemble2 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + + TestEnsembleListener listener1 = new TestEnsembleListener(); + TestEnsembleListener listener2 = new TestEnsembleListener(); + ensemble1.addEnsembleListener(listener1); + ensemble2.addEnsembleListener(listener2); + + ensemble1.afterPropertiesSet(); + ensemble1.start(); + ensemble2.afterPropertiesSet(); + ensemble2.start(); + + TestStateMachine stateMachine1 = new TestStateMachine(); + TestStateMachine stateMachine2 = new TestStateMachine(); + + ensemble1.join(stateMachine1); + ensemble2.join(stateMachine2); + assertThat(listener1.joinedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener2.joinedLatch.await(2, TimeUnit.SECONDS), is(true)); + + ensemble1.setState(new DefaultStateMachineContext(stateMachine1, "S1", "E1", null, null)); + assertThat(listener2.eventLatch.await(2, TimeUnit.SECONDS), is(true)); + } + + @Override + protected AnnotationConfigApplicationContext buildContext() { + return new AnnotationConfigApplicationContext(); + } + + private class TestEnsembleListener implements EnsembleListeger { + + volatile CountDownLatch joinedLatch = new CountDownLatch(1); + volatile CountDownLatch eventLatch = new CountDownLatch(1); + + @Override + public void stateMachineJoined(StateMachineContext context) { + joinedLatch.countDown(); + } + + @Override + public void stateMachineLeft(StateMachineContext context) { + } + + @Override + public void stateChanged(StateMachineContext context) { + eventLatch.countDown(); + } + + public void reset(int c1, int c2) { + joinedLatch = new CountDownLatch(c1); + eventLatch = new CountDownLatch(c2); + } + + } + + private class TestStateMachine implements StateMachine { + + @Override + public StateMachineAccessor getStateMachineAccessor() { + return null; + } + + @Override + public void start() { + } + + @Override + public void stop() { + } + + @Override + public boolean sendEvent(Message event) { + return false; + } + + @Override + public boolean sendEvent(String event) { + return false; + } + + @Override + public State getState() { + return null; + } + + @Override + public Collection> getStates() { + return null; + } + + @Override + public Collection> getTransitions() { + return null; + } + + @Override + public boolean isComplete() { + return false; + } + + @Override + public void addStateListener(StateMachineListener listener) { + } + + @Override + public void removeStateListener(StateMachineListener listener) { + } + + @Override + public State getInitialState() { + return null; + } + + @Override + public ExtendedState getExtendedState() { + return null; + } + + } + +} 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 new file mode 100644 index 00000000..bb5a734b --- /dev/null +++ b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java @@ -0,0 +1,222 @@ +/* + * 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.zookeeper; + +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.junit.Assert.assertThat; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.curator.framework.CuratorFramework; +import org.junit.Test; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.config.EnableStateMachine; +import org.springframework.statemachine.config.StateMachineConfigurerAdapter; +import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; +import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; +import org.springframework.statemachine.ensemble.DistributedStateMachine; +import org.springframework.statemachine.listener.StateMachineListenerAdapter; +import org.springframework.statemachine.state.State; +import org.springframework.statemachine.transition.Transition; + +public class ZookeeperStateMachineTests extends AbstractZookeeperTests { + + @Override + protected AnnotationConfigApplicationContext buildContext() { + return new AnnotationConfigApplicationContext(); + } + + @Test + @SuppressWarnings("unchecked") + public void testStateChanges() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class, Config1.class, Config2.class); + context.refresh(); + + StateMachine machine1 = + context.getBean("sm1", StateMachine.class); + StateMachine machine2 = + context.getBean("sm2", StateMachine.class); + + TestListener listener1 = new TestListener(); + TestListener listener2 = new TestListener(); + machine1.addStateListener(listener1); + machine2.addStateListener(listener2); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ZookeeperStateMachineEnsemble ensemble2 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ensemble1.afterPropertiesSet(); + ensemble2.afterPropertiesSet(); + ensemble1.start(); + ensemble2.start(); + + DistributedStateMachine machine1s = + new DistributedStateMachine(ensemble1, machine1); + + DistributedStateMachine machine2s = + new DistributedStateMachine(ensemble2, machine2); + + machine1s.afterPropertiesSet(); + machine2s.afterPropertiesSet(); + + machine1s.start(); + machine2s.start(); + + listener1.reset(1); + listener2.reset(1); + machine1s.sendEvent("E1"); + assertThat(listener1.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener1.stateChangedCount, is(1)); + assertThat(listener2.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener2.stateChangedCount, is(1)); + assertThat(machine1.getState().getIds(), containsInAnyOrder("S1")); + assertThat(machine2.getState().getIds(), containsInAnyOrder("S1")); + + listener1.reset(1); + listener2.reset(1); + machine1s.sendEvent("E2"); + assertThat(listener1.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener1.stateChangedCount, is(1)); + assertThat(listener2.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener2.stateChangedCount, is(1)); + assertThat(machine1.getState().getIds(), containsInAnyOrder("S2")); + assertThat(machine2.getState().getIds(), containsInAnyOrder("S2")); + } + + @Test + @SuppressWarnings("unchecked") + public void testJoinLaterShouldSyncState() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class, Config1.class, Config2.class); + context.refresh(); + + StateMachine machine1 = + context.getBean("sm1", StateMachine.class); + StateMachine machine2 = + context.getBean("sm2", StateMachine.class); + + TestListener listener1 = new TestListener(); + TestListener listener2 = new TestListener(); + machine1.addStateListener(listener1); + machine2.addStateListener(listener2); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ensemble1.afterPropertiesSet(); + ensemble1.start(); + + DistributedStateMachine machine1s = + new DistributedStateMachine(ensemble1, machine1); + + + machine1s.afterPropertiesSet(); + machine1s.start(); + + listener1.reset(1); + machine1s.sendEvent("E1"); + assertThat(listener1.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener1.stateChangedCount, is(1)); + assertThat(machine1.getState().getIds(), containsInAnyOrder("S1")); + + ZookeeperStateMachineEnsemble ensemble2 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + + ensemble2.afterPropertiesSet(); + ensemble2.start(); + DistributedStateMachine machine2s = + new DistributedStateMachine(ensemble2, machine2); + machine2s.afterPropertiesSet(); + machine2s.start(); + assertThat(machine2.getState().getIds(), containsInAnyOrder("S1")); + } + + @Configuration + @EnableStateMachine(name = "sm1") + static class Config1 extends SharedConfig { + } + + @Configuration + @EnableStateMachine(name = "sm2") + static class Config2 extends SharedConfig { + } + + static class SharedConfig extends StateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial("SI") + .state("S1") + .state("S2"); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions + .withExternal() + .source("SI") + .target("S1") + .event("E1") + .and() + .withExternal() + .source("S1") + .target("S2") + .event("E2"); + } + + } + + private static class TestListener extends StateMachineListenerAdapter { + + volatile CountDownLatch stateChangedLatch = new CountDownLatch(1); + volatile CountDownLatch transitionLatch = new CountDownLatch(0); + volatile int stateChangedCount = 0; + + @Override + public void stateChanged(State from, State to) { + stateChangedCount++; + stateChangedLatch.countDown(); + } + + @Override + public void transition(Transition transition) { + transitionLatch.countDown(); + } + + public void reset(int c1) { + reset(c1, 0); + } + + public void reset(int c1, int c2) { + stateChangedLatch = new CountDownLatch(c1); + transitionLatch = new CountDownLatch(c2); + stateChangedCount = 0; + } + + } + +} diff --git a/spring-statemachine-zookeeper/src/test/resources/log4j.properties b/spring-statemachine-zookeeper/src/test/resources/log4j.properties new file mode 100644 index 00000000..72283d00 --- /dev/null +++ b/spring-statemachine-zookeeper/src/test/resources/log4j.properties @@ -0,0 +1,8 @@ +log4j.rootCategory=INFO, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2} [%t] - %m%n + +log4j.category.org.springframework.statemachine.zookeeper=TRACE +