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

Add skeleton for distributed state machine

Browse Source 
- NOTE: not yet fully working distributed sm, this
  commit is considered to be a stage 1 of further
  commits on this matter.
- Preliminary concept of DistributedStateMachine,
  StateMachineEnsemble and StateMachinePersist.
- New module spring-statemachine-zookeeper order to
  persist state in a distributed matter.
- Refactoring concept of a state machine access to get
  better internal access into a sm via functional interfaces.
- Change build to use jdk7 and change test to rely on jdk8
  order to test functional lambdas.
- new zookeeper sample demonstrating distributed concepts.
- Relates t0 #35 and #34.
- No docs until things are fully implemented.
This commit is contained in:
Janne Valkealahti
2015-06-12 18:15:50 +01:00
parent 1648b39de1
commit c68fa5d09d
39 changed files with 2245 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
build.gradle
View File

@@ -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"

2
gradle.properties
View File

@@ -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

2
settings.gradle
View File

@@ -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') {

8
spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachine.java
View File

@@ -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<S, E> extends Region<S, E> {
*/
ExtendedState getExtendedState();
/**
* Gets the state machine accessor.
*
* @return the state machine accessor
*/
StateMachineAccessor<S, E> getStateMachineAccessor();
}

65
spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineContext.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public interface StateMachineContext<S, E> {
/**
* Gets the state machine.
*
* @return the state machine
*/
StateMachine<S, E> 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<String, Object> getEventHeaders();
/**
* Gets the extended state.
*
* @return the extended state
*/
ExtendedState getExtendedState();
}

27
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineAccess.java → spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java
View File

@@ -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<S, E> {
/**
* Execute given {@link StateMachineFunction} with all recursive regions.
*
* @param stateMachineAccess the state machine access
*/
void doWithAllRegions(StateMachineFunction<StateMachineAccess<S, E>> stateMachineAccess);
/**
* Sets the relay state machine.
*
@@ -42,4 +35,18 @@ public interface StateMachineAccess<S, E> {
*/
void setRelay(StateMachine<S, E> 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);
}

49
spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccessor.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public interface StateMachineAccessor<S, E> {
/**
* Execute given {@link StateMachineFunction} with all recursive regions.
*
* @param stateMachineAccess the state machine access
*/
void doWithAllRegions(StateMachineFunction<StateMachineAccess<S, E>> stateMachineAccess);
/**
* Gets all regions.
*
* @return the all regions
*/
List<StateMachineAccess<S, E>> withAllRegions();
}

4
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineFunction.java → spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineFunction.java
View File

@@ -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 <I> the function type
* @see StateMachineAccess
* @see StateMachineAccessor
*/
public interface StateMachineFunction<I> {

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

@@ -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<S, E> extends LifecycleObjectS
}
// set top-level machine as relay
final StateMachine<S, E> mm = machine;
((StateMachineAccess<S, E>)machine).doWithAllRegions(new StateMachineFunction<StateMachineAccess<S, E>>() {
final StateMachine<S, E> fmachine = machine;
fmachine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction<StateMachineAccess<S, E>>() {
@Override
public void apply(StateMachineAccess<S, E> stateMachineAccess) {
stateMachineAccess.setRelay(mm);
public void apply(StateMachineAccess<S, E> function) {
function.setRelay(fmachine);
}
});
});
return machine;
}

58
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/CompositeEnsembleListener.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public class CompositeEnsembleListener<S, E> extends AbstractCompositeListener<EnsembleListeger<S, E>> implements
EnsembleListeger<S, E> {
@Override
public void stateMachineJoined(StateMachineContext<S, E> context) {
for (Iterator<EnsembleListeger<S, E>> iterator = getListeners().reverse(); iterator.hasNext();) {
EnsembleListeger<S, E> listener = iterator.next();
listener.stateMachineJoined(context);
}
}
@Override
public void stateMachineLeft(StateMachineContext<S, E> context) {
for (Iterator<EnsembleListeger<S, E>> iterator = getListeners().reverse(); iterator.hasNext();) {
EnsembleListeger<S, E> listener = iterator.next();
listener.stateMachineLeft(context);
}
}
@Override
public void stateChanged(StateMachineContext<S, E> context) {
for (Iterator<EnsembleListeger<S, E>> iterator = getListeners().reverse(); iterator.hasNext();) {
EnsembleListeger<S, E> listener = iterator.next();
listener.stateChanged(context);
}
}
}

195
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implements StateMachine<S, E> {
private final static Log log = LogFactory.getLog(DistributedStateMachine.class);
private final String uuid = UUID.randomUUID().toString();
private final StateMachineEnsemble<S, E> ensemble;
private final StateMachine<S, E> delegate;
private final LocalEnsembleListener listener;
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<S, E> ensemble, StateMachine<S, E> 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<E> 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<S, E> getState() {
return delegate.getState();
}
@Override
public Collection<State<S, E>> getStates() {
return delegate.getStates();
}
@Override
public Collection<Transition<S, E>> getTransitions() {
return delegate.getTransitions();
}
@Override
public boolean isComplete() {
return delegate.isComplete();
}
@Override
public void addStateListener(StateMachineListener<S, E> listener) {
delegate.addStateListener(listener);
}
@Override
public void removeStateListener(StateMachineListener<S, E> listener) {
delegate.removeStateListener(listener);
}
@Override
public State<S, E> getInitialState() {
return delegate.getInitialState();
}
@Override
public ExtendedState getExtendedState() {
return delegate.getExtendedState();
}
@Override
public StateMachineAccessor<S, E> getStateMachineAccessor() {
return delegate.getStateMachineAccessor();
}
private class LocalStateMachineListener extends StateMachineListenerAdapter<S, E> {
@Override
public void stateChanged(StateContext<S, E> context) {
if (ObjectUtils.nullSafeEquals(uuid, context.getMessageHeader("uuid"))) {
ensemble.setState(new DefaultStateMachineContext<S, E>(delegate, context.getTransition().getTarget()
.getId(), context.getEvent(), context.getMessageHeaders(), context.getExtendedState()));
}
}
}
private class LocalEnsembleListener implements EnsembleListeger<S, E> {
@Override
public void stateMachineJoined(final StateMachineContext<S, E> 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<StateMachineAccess<S, E>>() {
@Override
public void apply(StateMachineAccess<S, E> 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<S, E> context) {
log.info("Requesting to stop delegating state machine " + delegate);
delegate.stop();
}
@Override
public void stateChanged(StateMachineContext<S, E> context) {
delegate.sendEvent(MessageBuilder.withPayload(context.getEvent()).copyHeaders(context.getEventHeaders()).build());
}
}
}

36
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public interface EnsembleListeger<S, E> {
void stateMachineJoined(StateMachineContext<S, E> context);
void stateMachineLeft(StateMachineContext<S, E> context);
void stateChanged(StateMachineContext<S, E> context);
}

67
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public interface StateMachineEnsemble<S, E> {
/**
* Request a join to a state machine ensemble.
*
* @param stateMachine the state machine
*/
void join(StateMachine<S, E> stateMachine);
/**
* Request a leave from an ensemble.
*
* @param stateMachine the state machine
*/
void leave(StateMachine<S, E> stateMachine);
/**
* Adds the ensemble listener.
*
* @param listener the listener
*/
void addEnsembleListener(EnsembleListeger<S, E> listener);
/**
* Removes the ensemble listener.
*
* @param listener the listener
*/
void removeEnsembleListener(EnsembleListeger<S, E> listener);
/**
* Sets the state.
*
* @param context the context
*/
void setState(StateMachineContext<S, E> context);
}

62
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public abstract class StateMachineEnsembleObjectSupport<S, E> extends LifecycleObjectSupport implements StateMachineEnsemble<S, E> {
private final CompositeEnsembleListener<S, E> ensembleListener = new CompositeEnsembleListener<S, E>();
@Override
public abstract void join(StateMachine<S, E> stateMachine);
@Override
public abstract void leave(StateMachine<S, E> stateMachine);
@Override
public void addEnsembleListener(EnsembleListeger<S, E> listener) {
ensembleListener.register(listener);
}
@Override
public void removeEnsembleListener(EnsembleListeger<S, E> listener) {
ensembleListener.unregister(listener);
}
protected void notifyJoined(StateMachineContext<S, E> context) {
ensembleListener.stateMachineJoined(context);
}
protected void notifyLeft(StateMachineContext<S, E> context) {
ensembleListener.stateMachineLeft(context);
}
protected void notifyStateChanged(StateMachineContext<S, E> context) {
ensembleListener.stateChanged(context);
}
}

47
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachinePersist.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public interface StateMachinePersist<S, E> {
/**
* Serialize a {@link StateMachineContext}.
*
* @param context the state machine context
* @return the serialized data
*/
byte[] serialize(StateMachineContext<S, E> context);
/**
* Deserialize a data into a {@link StateMachineContext}.
*
* @param data the data
* @return the state machine context
*/
StateMachineContext<S, E> deserialize(byte[] data);
}

9
spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/CompositeStateMachineListener.java
View File

@@ -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<S,E> extends AbstractCompositeListene
}
}
@Override
public void stateChanged(StateContext<S, E> context) {
for (Iterator<StateMachineListener<S, E>> iterator = getListeners().reverse(); iterator.hasNext();) {
StateMachineListener<S, E> listener = iterator.next();
listener.stateChanged(context);
}
}
@Override
public void stateEntered(State<S, E> state) {
for (Iterator<StateMachineListener<S, E>> iterator = getListeners().reverse(); iterator.hasNext();) {

8
spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListener.java
View File

@@ -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<S,E> {
*/
void stateChanged(State<S,E> from, State<S,E> to);
/**
* Notified when state is changed.
*
* @param context the state context
*/
void stateChanged(StateContext<S, E> context);
/**
* Notified when state is entered.
*

5
spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListenerAdapter.java
View File

@@ -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<S, E> implements StateMachineListener<S
public void stateChanged(State<S, E> from, State<S, E> to) {
}
@Override
public void stateChanged(StateContext<S, E> context) {
}
@Override
public void stateEntered(State<S, E> state) {
}

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

@@ -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<S, E> extends StateMachineObjectSuppo
private final Message<E> initialEvent;
private final ExtendedState extendedState;
private ExtendedState extendedState;
private volatile State<S,E> currentState;
@@ -165,6 +168,11 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
return extendedState;
}
@Override
public void setExtendedState(ExtendedState extendedState) {
this.extendedState = extendedState;
}
public void setHistoryState(PseudoState<S, E> history) {
this.history = history;
}
@@ -304,21 +312,46 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
@SuppressWarnings("unchecked")
@Override
public void doWithAllRegions(StateMachineFunction<StateMachineAccess<S, E>> stateMachineAccess) {
stateMachineAccess.apply(this);
for (State<S, E> state : states) {
if (state.isSubmachineState()) {
StateMachine<S, E> submachine = ((AbstractState<S, E>)state).getSubmachine();
if (submachine instanceof StateMachineAccess) {
((StateMachineAccess<S, E>)submachine).doWithAllRegions(stateMachineAccess);
}
} else if (state.isOrthogonal()) {
Collection<Region<S, E>> regions = ((AbstractState<S, E>)state).getRegions();
for (Region<S, E> region : regions) {
((StateMachineAccess<S, E>)region).doWithAllRegions(stateMachineAccess);
public StateMachineAccessor<S, E> getStateMachineAccessor() {
// TODO: needs cleaning and perhaps not an anonymous function
return new StateMachineAccessor<S, E>() {
@Override
public void doWithAllRegions(StateMachineFunction<StateMachineAccess<S, E>> stateMachineAccess) {
stateMachineAccess.apply(AbstractStateMachine.this);
for (State<S, E> state : states) {
if (state.isSubmachineState()) {
StateMachine<S, E> submachine = ((AbstractState<S, E>) state).getSubmachine();
submachine.getStateMachineAccessor().doWithAllRegions(stateMachineAccess);
} else if (state.isOrthogonal()) {
Collection<Region<S, E>> regions = ((AbstractState<S, E>) state).getRegions();
for (Region<S, E> region : regions) {
((StateMachine<S, E>)region).getStateMachineAccessor().doWithAllRegions(stateMachineAccess);
}
}
}
}
}
@Override
public List<StateMachineAccess<S, E>> withAllRegions() {
List<StateMachineAccess<S, E>> list = new ArrayList<StateMachineAccess<S, E>>();
list.add(AbstractStateMachine.this);
for (State<S, E> state : states) {
if (state.isSubmachineState()) {
StateMachine<S, E> submachine = ((AbstractState<S, E>) state).getSubmachine();
if (submachine instanceof StateMachineAccess) {
list.add((StateMachineAccess<S, E>)submachine);
}
} else if (state.isOrthogonal()) {
Collection<Region<S, E>> regions = ((AbstractState<S, E>) state).getRegions();
for (Region<S, E> region : regions) {
list.add((StateMachineAccess<S, E>) region);
}
}
}
return list;
}
};
}
@Override
@@ -353,6 +386,16 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
return buf.toString();
}
@Override
public void resetState(S state) {
for (State<S, E> s : getStates()) {
if (s.getId().equals(state)) {
currentState = s;
break;
}
}
}
protected boolean acceptEvent(Message<E> message) {
boolean accepted = currentState.sendEvent(message);
@@ -476,6 +519,8 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
}
entryToState(state, message, transition, stateMachine);
notifyStateChanged(notifyFrom, state);
StateContext<S, E> stateContext = buildStateContext(message, transition, stateMachine);
notifyStateChanged(stateContext);
} else if (currentState != null) {
if (findDeep != null) {
if (exit) {

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

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public class DefaultStateMachineContext<S, E> implements StateMachineContext<S, E> {
private final StateMachine<S, E> stateMachine;
private final S state;
private final E event;
private final Map<String, Object> 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<S, E> stateMachine, S state, E event, Map<String, Object> eventHeaders, ExtendedState extendedState) {
this.stateMachine = stateMachine;
this.state = state;
this.event = event;
this.eventHeaders = eventHeaders;
this.extendedState = extendedState;
}
@Override
public StateMachine<S, E> getStateMachine() {
return stateMachine;
}
@Override
public S getState() {
return state;
}
@Override
public E getEvent() {
return event;
}
@Override
public Map<String, Object> getEventHeaders() {
return eventHeaders;
}
@Override
public ExtendedState getExtendedState() {
return extendedState;
}
}

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

@@ -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<S, E> extends LifecycleObjectSup
}
}
protected void notifyStateChanged(StateContext<S, E> context) {
stateListener.stateChanged(context);
// if (contextEventsEnabled) {
// StateMachineEventPublisher eventPublisher = getStateMachineEventPublisher();
// if (eventPublisher != null) {
// eventPublisher.publishStateChanged(this, source, target);
// }
// }
}
protected void notifyStateEntered(State<S,E> state) {
stateListener.stateEntered(state);
if (contextEventsEnabled) {
@@ -188,6 +199,11 @@ public abstract class StateMachineObjectSupport<S, E> extends LifecycleObjectSup
stateChangedInRelay();
}
@Override
public void stateChanged(StateContext<S, E> context) {
stateListener.stateChanged(context);
}
@Override
public void stateEntered(State<S, E> state) {
stateListener.stateEntered(state);

154
spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java Normal file
View File

@@ -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<String, String> stateMachine = mock;
stateMachine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction<StateMachineAccess<String, String>>() {
@Override
public void apply(StateMachineAccess<String, String> function) {
function.setRelay(stateMachine);
}
});
assertThat(mock.relay, sameInstance(stateMachine));
}
@Test
public void testGetAllRegionsSetRelay() {
MockStateMachine mock = new MockStateMachine();
final StateMachine<String, String> stateMachine = mock;
stateMachine.getStateMachineAccessor().withAllRegions().stream()
.forEach(access -> access.setRelay(stateMachine));
assertThat(mock.relay, sameInstance(stateMachine));
}
private static class MockStateMachine implements StateMachine<String, String>, StateMachineAccess<String, String> {
StateMachine<String, String> relay;
@Override
public StateMachineAccessor<String, String> getStateMachineAccessor() {
return new StateMachineAccessor<String, String>() {
@Override
public void doWithAllRegions(StateMachineFunction<StateMachineAccess<String, String>> stateMachineAccess) {
stateMachineAccess.apply(MockStateMachine.this);
}
@Override
public List<StateMachineAccess<String, String>> withAllRegions() {
List<StateMachineAccess<String, String>> list = new ArrayList<StateMachineAccess<String,String>>();
list.add(MockStateMachine.this);
return list;
}
};
}
@Override
public void setRelay(StateMachine<String, String> 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<String> event) {
return false;
}
@Override
public boolean sendEvent(String event) {
return false;
}
@Override
public State<String, String> getState() {
return null;
}
@Override
public Collection<State<String, String>> getStates() {
return null;
}
@Override
public Collection<Transition<String, String>> getTransitions() {
return null;
}
@Override
public boolean isComplete() {
return false;
}
@Override
public void addStateListener(StateMachineListener<String, String> listener) {
}
@Override
public void removeStateListener(StateMachineListener<String, String> listener) {
}
@Override
public State<String, String> getInitialState() {
return null;
}
@Override
public ExtendedState getExtendedState() {
return null;
}
}
}

136
spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/DistributedStateMachineTests.java Normal file
View File

@@ -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<String, String> machine1 =
context.getBean("sm1", StateMachine.class);
StateMachine<String, String> machine2 =
context.getBean("sm2", StateMachine.class);
StateMachineEnsemble<String, String> ensemble = new InMemoryStateMachineEnsemble<String, String>();
DistributedStateMachine<String, String> machine1s =
new DistributedStateMachine<String, String>(ensemble, machine1);
DistributedStateMachine<String, String> machine2s =
new DistributedStateMachine<String, String>(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<String, String> machine1 =
context.getBean("sm1", StateMachine.class);
StateMachine<String, String> machine2 =
context.getBean("sm2", StateMachine.class);
StateMachineEnsemble<String, String> ensemble = new InMemoryStateMachineEnsemble<String, String>();
DistributedStateMachine<String, String> machine1s =
new DistributedStateMachine<String, String>(ensemble, machine1);
machine1s.afterPropertiesSet();
machine1s.start();
machine1s.sendEvent("E1");
assertThat(machine1.getState().getIds(), containsInAnyOrder("S1"));
DistributedStateMachine<String, String> machine2s =
new DistributedStateMachine<String, String>(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<String, String> {
@Override
public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
states
.withStates()
.initial("SI")
.state("S1")
.state("S2");
}
@Override
public void configure(StateMachineTransitionConfigurer<String, String> transitions) throws Exception {
transitions
.withExternal()
.source("SI")
.target("S1")
.event("E1")
.and()
.withExternal()
.source("S1")
.target("S2")
.event("E2");
}
}
}

51
spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java Normal file
View File

@@ -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<S, E> extends StateMachineEnsembleObjectSupport<S, E> {
private final Set<StateMachine<S, E>> joined = new HashSet<StateMachine<S,E>>();
private StateMachineContext<S, E> current;
@Override
public void join(StateMachine<S, E> stateMachine) {
if (!joined.contains(stateMachine)) {
joined.add(stateMachine);
notifyJoined(current);
}
}
@Override
public void leave(StateMachine<S, E> stateMachine) {
if (joined.remove(stateMachine)) {
notifyLeft(current);
}
}
@Override
public void setState(StateMachineContext<S, E> context) {
current = context;
notifyStateChanged(context);
}
}

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

@@ -131,6 +131,10 @@ public class ListenerTests extends AbstractStateMachineTests {
states.add(new Holder(from, to));
}
@Override
public void stateChanged(StateContext<TestStates, TestEvents> context) {
}
@Override
public void stateEntered(State<TestStates, TestEvents> state) {
}

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

@@ -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<Message<SpelEvents>> events = new ArrayList<Message<SpelEvents>>();
@Override
public StateMachineAccessor<SpelStates, SpelEvents> getStateMachineAccessor() {
return null;
}
@Override
public void start() {
}

4
spring-statemachine-samples/build.gradle
View File

@@ -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'
}

19
spring-statemachine-samples/zookeeper/.gitignore vendored Normal file
View File

@@ -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

121
spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/Application.java Normal file
View File

@@ -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<String, String> internalMachine;
@Bean
public StateMachine<String, String> stateMachine() throws Exception {
DistributedStateMachine<String, String> machine =
new DistributedStateMachine<String, String>(ensemble(), internalMachine);
return machine;
}
@Bean
public ZookeeperStateMachineEnsemble<String, String> ensemble() throws Exception {
ZookeeperStateMachineEnsemble<String, String> ensemble =
new ZookeeperStateMachineEnsemble<String, String>(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<String, String> {
@Override
public void configure(StateMachineStateConfigurer<String, String> states)
throws Exception {
states
.withStates()
.initial("LOCKED")
.state("UNLOCKED");
}
@Override
public void configure(StateMachineTransitionConfigurer<String, String> 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);
}
}

33
spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/StateMachineCommands.java Normal file
View File

@@ -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<String, String> {
@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";
}
}

8
spring-statemachine-samples/zookeeper/src/main/resources/META-INF/spring/spring-shell-plugin.xml Normal file
View File

@@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:component-scan base-package="demo" />
</beans>

19
spring-statemachine-samples/zookeeper/src/main/resources/statechartmodel.txt Normal file
View File

@@ -0,0 +1,19 @@
+----------------------------------------------------------------+
| SM |
+----------------------------------------------------------------+
| |
| +----------------+ +----------------+ |
| *-->| LOCKED | | UNLOCKED | |
| +----------------+ +----------------+ |
| +---| entry/ | | entry/ |---+ |
| | | exit/ | | exit/ | | |
| | | | | | | |
| PUSH| | |---COIN-->| | |COIN |
| | | | | | | |
| | | | | | | |
| | | |<--PUSH---| | | |
| +-->| | | |<--+ |
| | | | | |
| +----------------+ +----------------+ |
| |
+----------------------------------------------------------------+

92
spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersist.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public class KryoStateMachinePersist<S, E> implements StateMachinePersist<S, E> {
private static final ThreadLocal<Kryo> kryoThreadLocal = new ThreadLocal<Kryo>() {
@SuppressWarnings("rawtypes")
@Override
protected Kryo initialValue() {
Kryo kryo = new Kryo();
kryo.addDefaultSerializer(StateMachineContext.class, new StateMachineContextSerializer());
return kryo;
}
};
@Override
public byte[] serialize(StateMachineContext<S, E> 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<S, E> 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<S, E> extends Serializer<StateMachineContext<S, E>> {
@Override
public void write(Kryo kryo, Output output, StateMachineContext<S, E> context) {
kryo.writeClassAndObject(output, context.getEvent());
kryo.writeClassAndObject(output, context.getState());
}
@SuppressWarnings("unchecked")
@Override
public StateMachineContext<S, E> read(Kryo kryo, Input input, Class<StateMachineContext<S, E>> clazz) {
E event = (E) kryo.readClassAndObject(input);
S state = (S) kryo.readClassAndObject(input);
return new DefaultStateMachineContext<S, E>(null, state, event, null, null);
}
}
}

188
spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java Normal file
View File

@@ -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 <S> the type of state
* @param <E> the type of event
*/
public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObjectSupport<S, E> {
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<S, E> persist = new KryoStateMachinePersist<S, E>();
private final AtomicReference<StateWrapper> stateRef = new AtomicReference<StateWrapper>();
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<S, E> stateMachine) {
StateWrapper stateWrapper = stateRef.get();
if (stateWrapper == null) {
try {
StateWrapper currentStateWrapper = readCurrentContext();
stateRef.set(new StateWrapper(currentStateWrapper.context, currentStateWrapper.version));
stateWrapper = stateRef.get();
} catch (Exception e) {
log.error("Error reading current state during join", e);
}
}
notifyJoined(stateWrapper != null ? stateWrapper.context : null);
}
@Override
public void leave(StateMachine<S, E> stateMachine) {
}
@Override
public void setState(StateMachineContext<S, E> context) {
byte[] data = persist.serialize(context);
CuratorTransaction tx = curatorClient.inTransaction();
try {
Collection<CuratorTransactionResult> 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<S, E> 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<S, E> context;
private final int version;
public StateWrapper(StateMachineContext<S, E> context, int version) {
this.context = context;
this.version = version;
}
@Override
public String toString() {
return "StateWrapper [context=" + context + ", version=" + version + "]";
}
}
}

104
spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/AbstractZookeeperTests.java Normal file
View File

@@ -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();
}
}
}

40
spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersistTests.java Normal file
View File

@@ -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<String, String> persist = new KryoStateMachinePersist<String, String>();
StateMachineContext<String, String> contextOut =
new DefaultStateMachineContext<String, String>(null, "S1", "E1", null, null);
byte[] data = persist.serialize(contextOut);
StateMachineContext<String, String> contextIn = persist.deserialize(data);
assertThat(contextOut.getState(), is(contextIn.getState()));
assertThat(contextOut.getEvent(), is(contextIn.getEvent()));
}
}

212
spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java Normal file
View File

@@ -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<String, String> ensemble =
new ZookeeperStateMachineEnsemble<String, String>(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<String, String> ensemble =
new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
ensemble.afterPropertiesSet();
assertThat(curatorClient.checkExists().forPath("/foo/current"), notNullValue());
ensemble.setState(new DefaultStateMachineContext<String, String>(null, "S1","E1", null, null));
ensemble.setState(new DefaultStateMachineContext<String, String>(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<String, String> ensemble1 =
new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
ZookeeperStateMachineEnsemble<String, String> ensemble2 =
new ZookeeperStateMachineEnsemble<String, String>(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<String, String>(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<String, String> {
volatile CountDownLatch joinedLatch = new CountDownLatch(1);
volatile CountDownLatch eventLatch = new CountDownLatch(1);
@Override
public void stateMachineJoined(StateMachineContext<String, String> context) {
joinedLatch.countDown();
}
@Override
public void stateMachineLeft(StateMachineContext<String, String> context) {
}
@Override
public void stateChanged(StateMachineContext<String, String> context) {
eventLatch.countDown();
}
public void reset(int c1, int c2) {
joinedLatch = new CountDownLatch(c1);
eventLatch = new CountDownLatch(c2);
}
}
private class TestStateMachine implements StateMachine<String, String> {
@Override
public StateMachineAccessor<String, String> getStateMachineAccessor() {
return null;
}
@Override
public void start() {
}
@Override
public void stop() {
}
@Override
public boolean sendEvent(Message<String> event) {
return false;
}
@Override
public boolean sendEvent(String event) {
return false;
}
@Override
public State<String, String> getState() {
return null;
}
@Override
public Collection<State<String, String>> getStates() {
return null;
}
@Override
public Collection<Transition<String, String>> getTransitions() {
return null;
}
@Override
public boolean isComplete() {
return false;
}
@Override
public void addStateListener(StateMachineListener<String, String> listener) {
}
@Override
public void removeStateListener(StateMachineListener<String, String> listener) {
}
@Override
public State<String, String> getInitialState() {
return null;
}
@Override
public ExtendedState getExtendedState() {
return null;
}
}
}

222
spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java Normal file
View File

@@ -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<String, String> machine1 =
context.getBean("sm1", StateMachine.class);
StateMachine<String, String> 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<String, String> ensemble1 =
new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
ZookeeperStateMachineEnsemble<String, String> ensemble2 =
new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
ensemble1.afterPropertiesSet();
ensemble2.afterPropertiesSet();
ensemble1.start();
ensemble2.start();
DistributedStateMachine<String, String> machine1s =
new DistributedStateMachine<String, String>(ensemble1, machine1);
DistributedStateMachine<String, String> machine2s =
new DistributedStateMachine<String, String>(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<String, String> machine1 =
context.getBean("sm1", StateMachine.class);
StateMachine<String, String> 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<String, String> ensemble1 =
new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
ensemble1.afterPropertiesSet();
ensemble1.start();
DistributedStateMachine<String, String> machine1s =
new DistributedStateMachine<String, String>(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<String, String> ensemble2 =
new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
ensemble2.afterPropertiesSet();
ensemble2.start();
DistributedStateMachine<String, String> machine2s =
new DistributedStateMachine<String, String>(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<String, String> {
@Override
public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
states
.withStates()
.initial("SI")
.state("S1")
.state("S2");
}
@Override
public void configure(StateMachineTransitionConfigurer<String, String> 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<String, String> {
volatile CountDownLatch stateChangedLatch = new CountDownLatch(1);
volatile CountDownLatch transitionLatch = new CountDownLatch(0);
volatile int stateChangedCount = 0;
@Override
public void stateChanged(State<String, String> from, State<String, String> to) {
stateChangedCount++;
stateChangedLatch.countDown();
}
@Override
public void transition(Transition<String, String> 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;
}
}
}

8
spring-statemachine-zookeeper/src/test/resources/log4j.properties Normal file
View File

@@ -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