diff --git a/build.gradle b/build.gradle index c47d37c7..d95bfb5f 100644 --- a/build.gradle +++ b/build.gradle @@ -46,6 +46,7 @@ configure(allprojects) { repositories { mavenCentral() maven { url "http://repo.springsource.org/libs-release" } + maven { url "http://repo.springsource.org/libs-milestone" } } task integrationTest(type: Test) { @@ -194,6 +195,23 @@ project('spring-statemachine-redis') { } } +project('spring-statemachine-cluster') { + description = "Spring State Machine Cluster" + + dependencies { + compile project(":spring-statemachine-zookeeper") + compile "org.springframework.cloud:spring-cloud-cluster-zookeeper:$springCloudClusterVersion" + + testCompile project(":spring-statemachine-test") + 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(recipeProjects()) { dependencies { compile project(":spring-statemachine-recipes-common") diff --git a/gradle.properties b/gradle.properties index 06989a3b..98848d99 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,6 +1,7 @@ springShellVersion=1.1.0.RELEASE springSecurityVersion=4.0.3.RELEASE springDataRedisVersion=1.6.2.RELEASE +springCloudClusterVersion=1.0.0.RC1 jedisVersion=2.7.3 junitVersion=4.12 springVersion=4.2.4.RELEASE diff --git a/settings.gradle b/settings.gradle index 255c0750..42a66fc2 100644 --- a/settings.gradle +++ b/settings.gradle @@ -5,6 +5,7 @@ include 'spring-statemachine-test' include 'spring-statemachine-kryo' include 'spring-statemachine-zookeeper' include 'spring-statemachine-redis' +include 'spring-statemachine-cluster' include 'spring-statemachine-recipes' diff --git a/spring-statemachine-cluster/src/main/java/org/springframework/statemachine/cluster/LeaderZookeeperStateMachineEnsemble.java b/spring-statemachine-cluster/src/main/java/org/springframework/statemachine/cluster/LeaderZookeeperStateMachineEnsemble.java new file mode 100644 index 00000000..26ad1b8d --- /dev/null +++ b/spring-statemachine-cluster/src/main/java/org/springframework/statemachine/cluster/LeaderZookeeperStateMachineEnsemble.java @@ -0,0 +1,126 @@ +/* + * Copyright 2016 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.cluster; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.curator.framework.CuratorFramework; +import org.springframework.cloud.cluster.leader.Context; +import org.springframework.cloud.cluster.leader.DefaultCandidate; +import org.springframework.cloud.cluster.zk.leader.LeaderInitiator; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.ensemble.StateMachineEnsemble; +import org.springframework.statemachine.zookeeper.ZookeeperStateMachineEnsemble; + +/** + * {@link StateMachineEnsemble} backed by a zookeeper and leader functionality + * from a Spring Cloud Cluster. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public class LeaderZookeeperStateMachineEnsemble extends ZookeeperStateMachineEnsemble { + + private final Map, InitiatorHolder> holders = new HashMap<>(); + private final CuratorFramework curatorClient; + private final String basePath; + private StateMachine leader; + + /** + * Instantiates a new leader zookeeper state machine ensemble. + * + * @param curatorClient the curator client + * @param basePath the base zookeeper path + */ + public LeaderZookeeperStateMachineEnsemble(CuratorFramework curatorClient, String basePath) { + super(curatorClient, basePath); + this.curatorClient = curatorClient; + this.basePath = basePath; + } + + /** + * Instantiates a new leader zookeeper state machine ensemble. + * + * @param curatorClient the curator client + * @param basePath the base zookeeper path + * @param cleanState if true clean existing state + * @param logSize the log size + */ + public LeaderZookeeperStateMachineEnsemble(CuratorFramework curatorClient, String basePath, boolean cleanState, int logSize) { + super(curatorClient, basePath, cleanState, logSize); + this.curatorClient = curatorClient; + this.basePath = basePath; + } + + @Override + public void join(StateMachine stateMachine) { + super.join(stateMachine); + StateMachineCandidate candidate = new StateMachineCandidate(stateMachine); + LeaderInitiator initiator = new LeaderInitiator(curatorClient, candidate, basePath + "/leader"); + initiator.start(); + holders.put(stateMachine, new InitiatorHolder(candidate, initiator)); + } + + @Override + public void leave(StateMachine stateMachine) { + super.leave(stateMachine); + InitiatorHolder holder = holders.get(stateMachine); + holder.candidate.yieldLeadership(); + holder.initiator.stop(); + holders.remove(stateMachine); + } + + @Override + public StateMachine getLeader() { + return leader; + } + + private class InitiatorHolder { + final StateMachineCandidate candidate; + final LeaderInitiator initiator; + + public InitiatorHolder(StateMachineCandidate candidate, LeaderInitiator initiator) { + this.candidate = candidate; + this.initiator = initiator; + } + } + + private class StateMachineCandidate extends DefaultCandidate { + final StateMachine stateMachine; + + public StateMachineCandidate(StateMachine stateMachine) { + super(); + this.stateMachine = stateMachine; + } + + @Override + public void onGranted(Context ctx) { + super.onGranted(ctx); + leader = stateMachine; + notifyGranted(stateMachine); + } + + @Override + public void onRevoked(Context ctx) { + super.onRevoked(ctx); + leader = null; + notifyRevoked(stateMachine); + } + } +} diff --git a/spring-statemachine-cluster/src/test/java/org/springframework/statemachine/cluster/AbstractZookeeperTests.java b/spring-statemachine-cluster/src/test/java/org/springframework/statemachine/cluster/AbstractZookeeperTests.java new file mode 100644 index 00000000..cd9eb764 --- /dev/null +++ b/spring-statemachine-cluster/src/test/java/org/springframework/statemachine/cluster/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.cluster; + +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-cluster/src/test/java/org/springframework/statemachine/cluster/LeaderZookeeperStateMachineEnsembleTests.java b/spring-statemachine-cluster/src/test/java/org/springframework/statemachine/cluster/LeaderZookeeperStateMachineEnsembleTests.java new file mode 100644 index 00000000..2b94e50f --- /dev/null +++ b/spring-statemachine-cluster/src/test/java/org/springframework/statemachine/cluster/LeaderZookeeperStateMachineEnsembleTests.java @@ -0,0 +1,139 @@ +/* + * Copyright 2016 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.cluster; + +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.is; +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.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.config.EnableStateMachineFactory; +import org.springframework.statemachine.config.StateMachineConfigurerAdapter; +import org.springframework.statemachine.config.StateMachineFactory; +import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer; +import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; +import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; +import org.springframework.statemachine.ensemble.EnsembleListenerAdapter; +import org.springframework.statemachine.ensemble.StateMachineEnsemble; + +public class LeaderZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { + + @Override + protected AnnotationConfigApplicationContext buildContext() { + return new AnnotationConfigApplicationContext(); + } + + @Test + @SuppressWarnings("unchecked") + public void testLeader() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class, Config1.class); + context.refresh(); + + StateMachineFactory factory = context.getBean(StateMachineFactory.class); + StateMachineEnsemble stateMachineEnsemble = context.getBean(StateMachineEnsemble.class); + TestEnsembleListener listener = context.getBean(TestEnsembleListener.class); + + StateMachine machine1 = factory.getStateMachine(); + assertThat(machine1.getState().getIds(), contains("S1")); + assertThat(listener.latch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(stateMachineEnsemble.getLeader(), is(machine1)); + + listener.reset(1); + StateMachine machine2 = factory.getStateMachine(); + stateMachineEnsemble.leave(machine1); + assertThat(listener.latch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(stateMachineEnsemble.getLeader(), is(machine2)); + } + + @Configuration + @EnableStateMachineFactory + static class Config1 extends StateMachineConfigurerAdapter { + + @Autowired + private CuratorFramework curatorClient; + + @Override + public void configure(StateMachineConfigurationConfigurer config) throws Exception { + config + .withConfiguration() + .autoStartup(true) + .and() + .withDistributed() + .ensemble(stateMachineEnsemble()); + } + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial("S1") + .state("S2") + .state("S3"); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions + .withExternal() + .source("S1").target("S2") + .event("E1") + .and() + .withExternal() + .source("S2").target("S3") + .event("E2") + .and() + .withExternal() + .source("S3").target("S1") + .event("E3"); + } + + @Bean + public StateMachineEnsemble stateMachineEnsemble() throws Exception { + LeaderZookeeperStateMachineEnsemble ensemble = new LeaderZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ensemble.addEnsembleListener(testEnsembleListener()); + return ensemble; + } + + @Bean + public TestEnsembleListener testEnsembleListener() { + return new TestEnsembleListener(); + } + + } + + static class TestEnsembleListener extends EnsembleListenerAdapter { + CountDownLatch latch = new CountDownLatch(1); + + @Override + public void ensembleLeaderGranted(StateMachine stateMachine) { + latch.countDown(); + } + + void reset(int a1) { + latch = new CountDownLatch(a1); + } + } + +} diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/action/DistributedLeaderAction.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/action/DistributedLeaderAction.java new file mode 100644 index 00000000..e2685027 --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/action/DistributedLeaderAction.java @@ -0,0 +1,71 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.statemachine.action; + +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.ensemble.StateMachineEnsemble; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +/** + * {@link Action} which is used to wrap execution of an {@link Action} + * so that only a {@link StateMachine} considered to be a leader in an + * {@link StateMachineEnsemble} will do the execution. + * + * Executing action via {@code DistributedLeaderAction} is bound to if + * current machine is a leader or not. Effectively this means that if + * leader doesn't exist, execution is discarded. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public class DistributedLeaderAction implements Action { + + private final static Log log = LogFactory.getLog(DistributedLeaderAction.class); + private final Action action; + private final StateMachineEnsemble ensemble; + + /** + * Instantiates a new distributed leader action. + * + * @param action the action + * @param ensemble the ensemble + */ + public DistributedLeaderAction(Action action, StateMachineEnsemble ensemble) { + Assert.notNull(action, "Action must be set"); + Assert.notNull(ensemble, "Ensemble must be set"); + this.action = action; + this.ensemble = ensemble; + } + + @Override + public void execute(StateContext context) { + StateMachine left = context.getStateMachine(); + StateMachine right = ensemble.getLeader(); + boolean leader = (left != null & right != null) && ObjectUtils.nullSafeEquals(left.getUuid(), right.getUuid()); + if (log.isDebugEnabled()) { + log.debug("Distibuted action leader status is " + leader); + } + if (leader) { + action.execute(context); + } + } +} 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 index 0a926100..61f5b8e5 100644 --- 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 @@ -64,4 +64,19 @@ public class CompositeEnsembleListener extends AbstractCompositeListener stateMachine) { + for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { + EnsembleListener listener = iterator.next(); + listener.ensembleLeaderGranted(stateMachine); + } + } + + @Override + public void ensembleLeaderRevoked(StateMachine stateMachine) { + for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { + EnsembleListener listener = iterator.next(); + listener.ensembleLeaderRevoked(stateMachine); + } + } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java index e583891b..81385b94 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java @@ -312,6 +312,13 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem throw exception; } + @Override + public void ensembleLeaderGranted(StateMachine stateMachine) { + } + + @Override + public void ensembleLeaderRevoked(StateMachine stateMachine) { + } } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListener.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListener.java index fe709687..17678eba 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListener.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListener.java @@ -69,4 +69,19 @@ public interface EnsembleListener { */ void ensembleError(StateMachineEnsembleException exception); + /** + * Called when a state machine is granted a leader role + * in an ensemble. + * + * @param stateMachine the state machine + */ + void ensembleLeaderGranted(StateMachine stateMachine); + + /** + * Called when a state machine is revoked from a leader role + * in an ensemble. + * + * @param stateMachine the state machine + */ + void ensembleLeaderRevoked(StateMachine stateMachine); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListenerAdapter.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListenerAdapter.java new file mode 100644 index 00000000..de71165a --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListenerAdapter.java @@ -0,0 +1,55 @@ +/* + * Copyright 2016 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; + +/** + * Adapter implementation of {@link EnsembleListener} implementing all + * methods which extended implementation can override. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public class EnsembleListenerAdapter implements EnsembleListener { + + @Override + public void stateMachineJoined(StateMachine stateMachine, StateMachineContext context) { + } + + @Override + public void stateMachineLeft(StateMachine stateMachine, StateMachineContext context) { + } + + @Override + public void stateChanged(StateMachineContext context) { + } + + @Override + public void ensembleError(StateMachineEnsembleException exception) { + } + + @Override + public void ensembleLeaderGranted(StateMachine stateMachine) { + } + + @Override + public void ensembleLeaderRevoked(StateMachine stateMachine) { + } +} diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java index 0bfb23ac..a994f1d0 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -80,4 +80,12 @@ public interface StateMachineEnsemble { */ StateMachineContext getState(); + /** + * Gets the ensemble leader. If returned machine + * is {@code NULL} it indicates that this ensemble + * doesn't know any leader. + * + * @return the ensemble leader + */ + StateMachine getLeader(); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java index 9e95d804..eb98f315 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java @@ -63,11 +63,18 @@ public abstract class StateMachineEnsembleObjectSupport extends LifecycleO ensembleListener.ensembleError(exception); } + protected void notifyGranted(StateMachine stateMachine) { + ensembleListener.ensembleLeaderGranted(stateMachine); + } + + protected void notifyRevoked(StateMachine stateMachine) { + ensembleListener.ensembleLeaderRevoked(stateMachine); + } + protected void notifyStateChanged(StateMachineContext context) { if (log.isTraceEnabled()) { log.trace("Notify notifyStateChanged " + context); } ensembleListener.stateChanged(context); } - } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/DistributedLeaderActionTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/DistributedLeaderActionTests.java new file mode 100644 index 00000000..5df67e58 --- /dev/null +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/DistributedLeaderActionTests.java @@ -0,0 +1,142 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.statemachine.action; + +import static org.hamcrest.Matchers.is; +import static org.junit.Assert.assertThat; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.statemachine.AbstractStateMachineTests; +import org.springframework.statemachine.StateContext; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.config.EnableStateMachineFactory; +import org.springframework.statemachine.config.StateMachineConfigurerAdapter; +import org.springframework.statemachine.config.StateMachineFactory; +import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer; +import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; +import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; +import org.springframework.statemachine.ensemble.InMemoryStateMachineEnsemble; +import org.springframework.statemachine.ensemble.StateMachineEnsemble; + +public class DistributedLeaderActionTests extends AbstractStateMachineTests { + + @Override + protected AnnotationConfigApplicationContext buildContext() { + return new AnnotationConfigApplicationContext(); + } + + @Test + public void testAction() throws Exception { + context.register(Config1.class); + context.refresh(); + + TestLeaderAction action = context.getBean("testLeaderAction", TestLeaderAction.class); + + @SuppressWarnings("unchecked") + StateMachineFactory factory = context.getBean(StateMachineFactory.class); + + StateMachine machine1 = factory.getStateMachine(); + StateMachine machine2 = factory.getStateMachine(); + machine1.sendEvent("E1"); + assertThat(action.latch.await(1, TimeUnit.SECONDS), is(false)); + assertThat(action.count, is(1)); + action.reset(2); + + machine2.sendEvent("E2"); + assertThat(action.latch.await(1, TimeUnit.SECONDS), is(false)); + assertThat(action.count, is(1)); + } + + @Configuration + @EnableStateMachineFactory + static class Config1 extends StateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineConfigurationConfigurer config) throws Exception { + config + .withConfiguration() + .autoStartup(true) + .and() + .withDistributed() + .ensemble(ensemble()); + } + + @Override + public void configure(StateMachineStateConfigurer states) throws Exception { + states + .withStates() + .initial("S1") + .state("S2", testDistributedLeaderAction(), null) + .state("S3", testDistributedLeaderAction(), null); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) throws Exception { + transitions + .withExternal() + .source("S1").target("S2") + .event("E1") + .and() + .withExternal() + .source("S2").target("S3") + .event("E2") + .and() + .withExternal() + .source("S3").target("S1") + .event("E3"); + } + + @Bean + public StateMachineEnsemble ensemble() { + return new InMemoryStateMachineEnsemble(); + } + + @Bean + public Action testLeaderAction() { + return new TestLeaderAction(); + } + + @Bean + public Action testDistributedLeaderAction() { + return new DistributedLeaderAction<>(testLeaderAction(), ensemble()); + } + + } + + static class TestLeaderAction implements Action { + + CountDownLatch latch = new CountDownLatch(2); + int count = 0; + + @Override + public void execute(StateContext context) { + count++; + latch.countDown(); + } + + public void reset(int a1) { + count = 0; + latch = new CountDownLatch(a1); + } + } + +} diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java index 2e278bf4..05d1334a 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -24,13 +24,16 @@ import org.springframework.statemachine.StateMachineContext; public class InMemoryStateMachineEnsemble extends StateMachineEnsembleObjectSupport { private final Set> joined = new HashSet>(); - + private StateMachine leader; private StateMachineContext current; @Override public void join(StateMachine stateMachine) { if (!joined.contains(stateMachine)) { joined.add(stateMachine); + if (leader == null) { + leader = stateMachine; + } notifyJoined(stateMachine, current); } } @@ -38,6 +41,14 @@ public class InMemoryStateMachineEnsemble extends StateMachineEnsembleObje @Override public void leave(StateMachine stateMachine) { if (joined.remove(stateMachine)) { + if (leader == stateMachine) { + StateMachine newLeader = null; + try { + newLeader = joined.iterator().next(); + } catch (Exception e) { + } + leader = newLeader; + } notifyLeft(stateMachine, current); } } @@ -53,4 +64,8 @@ public class InMemoryStateMachineEnsemble extends StateMachineEnsembleObje return current; } + @Override + public StateMachine getLeader() { + return leader; + } } diff --git a/spring-statemachine-samples/build.gradle b/spring-statemachine-samples/build.gradle index b9f3eab4..5cbbfdfa 100644 --- a/spring-statemachine-samples/build.gradle +++ b/spring-statemachine-samples/build.gradle @@ -56,7 +56,7 @@ project('spring-statemachine-samples-persist') { project('spring-statemachine-samples-web') { description = 'Spring State Machine Web Sample' dependencies { - compile project(":spring-statemachine-zookeeper") + compile project(":spring-statemachine-cluster") compile("org.springframework.boot:spring-boot-starter-security:$springBootVersion") compile("org.springframework.boot:spring-boot-starter-web:$springBootVersion") compile("org.springframework.boot:spring-boot-starter-websocket:$springBootVersion") diff --git a/spring-statemachine-samples/web/src/main/java/demo/web/StateMachineConfig.java b/spring-statemachine-samples/web/src/main/java/demo/web/StateMachineConfig.java index 0ea26409..ddb277d9 100644 --- a/spring-statemachine-samples/web/src/main/java/demo/web/StateMachineConfig.java +++ b/spring-statemachine-samples/web/src/main/java/demo/web/StateMachineConfig.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -22,10 +22,14 @@ import org.apache.commons.logging.LogFactory; 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.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.simp.SimpMessagingTemplate; import org.springframework.statemachine.StateContext; import org.springframework.statemachine.action.Action; +import org.springframework.statemachine.action.DistributedLeaderAction; +import org.springframework.statemachine.cluster.LeaderZookeeperStateMachineEnsemble; import org.springframework.statemachine.config.EnableStateMachine; import org.springframework.statemachine.config.EnumStateMachineConfigurerAdapter; import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer; @@ -33,7 +37,6 @@ import org.springframework.statemachine.config.builders.StateMachineStateConfigu import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; import org.springframework.statemachine.ensemble.StateMachineEnsemble; import org.springframework.statemachine.guard.Guard; -import org.springframework.statemachine.zookeeper.ZookeeperStateMachineEnsemble; @Configuration public class StateMachineConfig { @@ -45,8 +48,12 @@ public class StateMachineConfig { static class Config extends EnumStateMachineConfigurerAdapter { + @Autowired + private SimpMessagingTemplate simpMessagingTemplate; + @Override - public void configure(StateMachineConfigurationConfigurer config) throws Exception { + public void configure(StateMachineConfigurationConfigurer config) + throws Exception { config .withDistributed() .ensemble(stateMachineEnsemble()) @@ -72,7 +79,7 @@ public class StateMachineConfig { .parent(States.S1) .initial(States.S11) .state(States.S11) - .state(States.S12) + .state(States.S12, distAction(), null) .and() .withStates() .parent(States.S0) @@ -178,9 +185,14 @@ public class StateMachineConfig { return new SetVariableAction(); } + @Bean + public DistributedLeaderAction distAction() throws Exception { + return new DistributedLeaderAction<>(new HelloAction(simpMessagingTemplate), stateMachineEnsemble()); + } + @Bean public StateMachineEnsemble stateMachineEnsemble() throws Exception { - return new ZookeeperStateMachineEnsemble(curatorClient(), "/foo"); + return new LeaderZookeeperStateMachineEnsemble(curatorClient(), "/foo"); } @Bean @@ -193,7 +205,6 @@ public class StateMachineConfig { client.start(); return client; } - } public enum States { @@ -247,7 +258,22 @@ public class StateMachineConfig { context.getExtendedState().getVariables().put("testVariable", testVariable); } } - } + private static class HelloAction implements Action { + + SimpMessagingTemplate simpMessagingTemplate; + + public HelloAction(SimpMessagingTemplate simpMessagingTemplate) { + this.simpMessagingTemplate = simpMessagingTemplate; + } + + @Override + public void execute(StateContext context) { + log.info("Hello"); + StateMachineMessage message = new StateMachineMessage(); + message.setMessage("Hello action"); + simpMessagingTemplate.convertAndSend("/topic/sm.message", message); + } + } } diff --git a/spring-statemachine-samples/web/src/main/java/demo/web/StateMachineController.java b/spring-statemachine-samples/web/src/main/java/demo/web/StateMachineController.java index 4a27aa1e..5033d40b 100644 --- a/spring-statemachine-samples/web/src/main/java/demo/web/StateMachineController.java +++ b/spring-statemachine-samples/web/src/main/java/demo/web/StateMachineController.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -34,6 +34,8 @@ import org.springframework.messaging.simp.annotation.SubscribeMapping; import org.springframework.messaging.support.MessageBuilder; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.StateMachineException; +import org.springframework.statemachine.ensemble.EnsembleListenerAdapter; +import org.springframework.statemachine.ensemble.StateMachineEnsemble; import org.springframework.statemachine.listener.StateMachineListenerAdapter; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; @@ -59,6 +61,9 @@ public class StateMachineController { @Autowired private StateMachine stateMachine; + @Autowired + private StateMachineEnsemble stateMachineEnsemble; + @PostConstruct public void setup() { @@ -102,6 +107,28 @@ public class StateMachineController { } }); + + stateMachineEnsemble.addEnsembleListener(new EnsembleListenerAdapter() { + + @Override + public void ensembleLeaderGranted(StateMachine stateMachine) { + StateMachineMessage message = new StateMachineMessage(); + message.setMessage("Leader granted " + stateMachine.getUuid().toString()); + simpMessagingTemplate.convertAndSend("/topic/sm.message", message); + } + + @Override + public void ensembleLeaderRevoked(StateMachine stateMachine) { + StateMachineMessage message = new StateMachineMessage(); + message.setMessage("Leader revoked " + stateMachine.getUuid().toString()); + simpMessagingTemplate.convertAndSend("/topic/sm.message", message); + } + }); + } + + @SubscribeMapping("/sm.uuid") + public String retrieveUuid() { + return stateMachine.getUuid().toString(); } @SubscribeMapping("/sm.states") @@ -123,12 +150,24 @@ public class StateMachineController { public void sendEvent(@RequestParam(value = "id") Events id, @RequestParam(value = "testVariable", required = false) String testVariable) { log.info("Got request to send event " + id + " testVariable " + testVariable); - Message message = MessageBuilder - .withPayload(id) - .setHeader("testVariable", testVariable) - .build(); - stateMachine.sendEvent(message); - } + Message message = MessageBuilder + .withPayload(id) + .setHeader("testVariable", testVariable) + .build(); + stateMachine.sendEvent(message); + } + + @RequestMapping("/join") + @ResponseStatus(HttpStatus.OK) + public void joinEnsemble() { + stateMachineEnsemble.join(stateMachine); + } + + @RequestMapping("/leave") + @ResponseStatus(HttpStatus.OK) + public void leaveEnsemble() { + stateMachineEnsemble.leave(stateMachine); + } @RequestMapping(value = "/states", method = RequestMethod.GET, produces="application/json") @ResponseBody diff --git a/spring-statemachine-samples/web/src/main/resources/statechartmodel.txt b/spring-statemachine-samples/web/src/main/resources/statechartmodel.txt index ec8eba11..62f0269f 100644 --- a/spring-statemachine-samples/web/src/main/resources/statechartmodel.txt +++ b/spring-statemachine-samples/web/src/main/resources/statechartmodel.txt @@ -1,44 +1,44 @@ -+----------------------------------------------------------------------------------------------+ -| S0 | -+----------------------------------------------------------------------------------------------+ -| entry/ | -| exit/ | -| H/[foo.equals(0)]; | -| | -| +-------------------------+ +--------------------------------------------+ | -| *-->| S1 | | S2 | | -| +-------------------------+ +--------------------------------------------+ | -| | entry/ | C | entry/ | | -| D | exit/ |----->| exit/ | | -|<-----------| H/ | | H/[foo.equals(1)]; | | -| | | | | | -| | +---------------+ | K | +------------------------------+ | | -| | *-->| S11 | |<-----| *-->| S21 | | | -| | +---------------+ | | +------------------------------+ | | -| | | entry/ | | F | | entry/ | | | -| | | exit/ |<---------| | exit/ | | | -| | | | | | | +--------------+ | | | -| | B | | | | | *-->| s211 | | | | -| |---->| J | | | F | +--------------+ G | | | -| | | +-------+ | |-------------------->| entry/ |----------------->| -| | +--| | | | | | G | | exit/ | | | | -| | | | | v |------------------------>| | | E | | -| | | +---------------+ | | | B | |<-----------------| -| | | | | |------>| | | | | -| | | +---------------+ | | | | | D | | | -| | I| | S12 | | | | +--| |------>| | | -| | | +---------------+ | | | | +--------------+ | | | -| | | | entry/ | | | | | | | | -| | | | exit/ | | | | I| +--------------+ | | | -| | | | | | | | | | s212 | | | | -| | +->| | | | | | +--------------+ | | | -| +--| | | | | | +->| entry/ | | | | -| | | | | | I | | | exit/ | | | | -| | | | |------------------------>| | | | | -| A| | | | | | | +--------------+ | | | -| | | | | | | | | | | -| | | +---------------+ | | +------------------------------+ | | -| +->| | | | | -| +-------------------------+ +--------------------------------------------+ | -| A[foo.equals(1)]; | -+----------------------------------------------------------------------------------------------+ ++-----------------------------------------------------------------------------------------------+ +| S0 | ++-----------------------------------------------------------------------------------------------+ +| entry/ | +| exit/ | +| H/[foo.equals(0)]; | +| | +| +--------------------------+ +--------------------------------------------+ | +| *-->| S1 | | S2 | | +| +--------------------------+ +--------------------------------------------+ | +| | entry/ | C | entry/ | | +| D | exit/ |----->| exit/ | | +|<-----------| H/ | | H/[foo.equals(1)]; | | +| | | | | | +| | +----------------+ | K | +------------------------------+ | | +| | *-->| S11 | |<-----| *-->| S21 | | | +| | +----------------+ | | +------------------------------+ | | +| | | entry/ | | F | | entry/ | | | +| | | exit/ |<---------| | exit/ | | | +| | | | | | | +--------------+ | | | +| | B | | | | | *-->| s211 | | | | +| |---->| J | | | F | +--------------+ G | | | +| | | +-------+ | |-------------------->| entry/ |----------------->| +| | +--| | | | | | G | | exit/ | | | | +| | | | | v |------------------------>| | | E | | +| | | +----------------+ | | | B | |<-----------------| +| | | | | |------>| | | | | +| | | +----------------+ | | | | | D | | | +| | I| | S12 | | | | +--| |------>| | | +| | | +----------------+ | | | | +--------------+ | | | +| | | | entry/ | | | | | | | | +| | | | distAction() | | | | I| +--------------+ | | | +| | | | exit/ | | | | | | s212 | | | | +| | +->| | | | | | +--------------+ | | | +| +--| | | | | | +->| entry/ | | | | +| | | | | | I | | | exit/ | | | | +| | | | |------------------------>| | | | | +| A| | | | | | | +--------------+ | | | +| | | | | | | | | | | +| | | +----------------+ | | +------------------------------+ | | +| +->| | | | | +| +--------------------------+ +--------------------------------------------+ | +| A[foo.equals(1)]; | ++-----------------------------------------------------------------------------------------------+ diff --git a/spring-statemachine-samples/web/src/main/resources/static/index.html b/spring-statemachine-samples/web/src/main/resources/static/index.html index 9b4a0d3b..5ec2c733 100755 --- a/spring-statemachine-samples/web/src/main/resources/static/index.html +++ b/spring-statemachine-samples/web/src/main/resources/static/index.html @@ -1,15 +1,15 @@ - - Spring Statemachine Zookeeper Demo - + + Spring Statemachine Zookeeper Demo + - - - + + + - + @@ -23,18 +23,18 @@
-
- - - - - - - - - - -
+
+ + + + + + + + + + +
@@ -42,44 +42,53 @@
-
-
-

States

- -
-
-

Variables

- -
-
-

Messages

-
- -
-
-
- - +
+
+ + +
+
+
+
UUID: {{uuid}}
+
+
+
+

States

+ +
+
+

Variables

+ +
+
+

Messages

+
+ +
+
+
+ + - - - - - + + + + + diff --git a/spring-statemachine-samples/web/src/main/resources/static/js/controllers.js b/spring-statemachine-samples/web/src/main/resources/static/js/controllers.js index c6e3d650..58c00953 100644 --- a/spring-statemachine-samples/web/src/main/resources/static/js/controllers.js +++ b/spring-statemachine-samples/web/src/main/resources/static/js/controllers.js @@ -7,10 +7,11 @@ angular.module('springChat.controllers', ['toaster']) var typing = undefined; + $scope.uuid = ''; $scope.participants = []; $scope.variables = []; - $scope.messages = []; - $scope.newMessage = ''; + $scope.messages = []; + $scope.newMessage = ''; $scope.sendEvent = function(event) { $http.post('/event', null, {params:{"id": event}}). @@ -24,11 +25,27 @@ angular.module('springChat.controllers', ['toaster']) }); }; + $scope.joinEnsemble = function(event) { + $http.post('/join', null, {}). + success(function(data) { + }); + }; + + $scope.leaveEnsemble = function(event) { + $http.post('/leave', null, {}). + success(function(data) { + }); + }; + var initStompClient = function() { chatSocket.init('/ws'); chatSocket.connect(function(frame) { + chatSocket.subscribe("/app/sm.uuid", function(message) { + $scope.uuid = message.body; + }); + chatSocket.subscribe("/app/sm.states", function(message) { $scope.participants = JSON.parse(message.body); }); @@ -55,16 +72,16 @@ angular.module('springChat.controllers', ['toaster']) chatSocket.subscribe("/topic/sm.message", function(message) { $scope.messages.unshift(JSON.parse(message.body)); - }); + }); chatSocket.subscribe("/user/queue/errors", function(message) { toaster.pop('error', "Error", message.body); - }); + }); }, function(error) { toaster.pop('error', 'Error', 'Connection error ' + error); - }); + }); }; initStompClient(); diff --git a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java index d2b7a09b..c87abb89 100644 --- a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java +++ b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -167,6 +167,11 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj } } + @Override + public StateMachine getLeader() { + return null; + } + private void joinQueued() { StateMachine stateMachine = null; synchronized (joinLock) { diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java index 7a48571f..78712031 100644 --- a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java +++ b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java @@ -671,6 +671,14 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { errorLatch.countDown(); } + @Override + public void ensembleLeaderGranted(StateMachine stateMachine) { + } + + @Override + public void ensembleLeaderRevoked(StateMachine stateMachine) { + } + public void reset(int c1, int c2) { reset(c1, c2, 0); }