From 34a28b57fa49ed0a0be11a2e3a2738fa3f3ec95a Mon Sep 17 00:00:00 2001 From: xJoeWoo Date: Tue, 16 Jun 2020 15:52:52 +0800 Subject: [PATCH] Add generic thread safe support to PersistStateMachineHandler --- .../AbstractPersistStateMachineHandler.java | 149 ++++++++++++++++++ .../FactoryPersistStateMachineHandler.java | 50 ++++++ .../GenericPersistStateMachineHandler.java | 53 +++++++ .../persist/PersistStateMachineHandler.java | 137 ++-------------- .../PersistStateMachineHandlerTests.java | 61 ++++++- 5 files changed, 316 insertions(+), 134 deletions(-) create mode 100644 spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/AbstractPersistStateMachineHandler.java create mode 100644 spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/FactoryPersistStateMachineHandler.java create mode 100644 spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/GenericPersistStateMachineHandler.java diff --git a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/AbstractPersistStateMachineHandler.java b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/AbstractPersistStateMachineHandler.java new file mode 100644 index 00000000..9e7e658b --- /dev/null +++ b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/AbstractPersistStateMachineHandler.java @@ -0,0 +1,149 @@ +/* + * Copyright 2015-2019 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 + * + * https://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.recipes.persist; + +import org.springframework.messaging.Message; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.access.StateMachineAccess; +import org.springframework.statemachine.listener.AbstractCompositeListener; +import org.springframework.statemachine.state.State; +import org.springframework.statemachine.support.DefaultStateMachineContext; +import org.springframework.statemachine.support.LifecycleObjectSupport; +import org.springframework.statemachine.support.StateMachineInterceptorAdapter; +import org.springframework.statemachine.transition.Transition; +import reactor.core.publisher.Mono; + +import java.util.Iterator; +import java.util.List; + +/** + * {@code AbstractPersistStateMachineHandler} is a base recipe which can be used to + * handle a state change of an arbitrary entity in a persistent storage. + * + * @author Janne Valkealahti + */ +public abstract class AbstractPersistStateMachineHandler extends LifecycleObjectSupport { + + protected final PersistingStateChangeInterceptor interceptor = new PersistingStateChangeInterceptor(); + protected final CompositePersistStateChangeListener listeners = new CompositePersistStateChangeListener(); + + protected abstract StateMachine getInitStateMachine(); + + protected void initStateMachine(StateMachine stateMachine) { + List> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions(); + for (StateMachineAccess a : withAllRegions) { + a.addStateMachineInterceptor(interceptor); + } + } + + /** + * Handle event with entity. + * + * @param event the event + * @param state the state + * @return true if event was accepted + * @see #handleEventWithStateReactively(Message, Object) + */ + @Deprecated + public boolean handleEventWithState(Message event, S state) { + StateMachine stateMachine = getInitStateMachine(); + stateMachine.stopReactively().block(); + List> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions(); + for (StateMachineAccess a : withAllRegions) { + a.resetStateMachine(new DefaultStateMachineContext(state, null, null, null)); + } + stateMachine.startReactively().block(); + return stateMachine.sendEvent(event); + } + + /** + * Handle event with entity reactively. + * + * @param event the event + * @param state the state + * @return mono for completion + */ + public Mono handleEventWithStateReactively(Message event, S state) { + StateMachine stateMachine = getInitStateMachine(); + // TODO: REACTOR add docs and revisit this function concept + return Mono.from(stateMachine.stopReactively()) + .then(Mono.fromRunnable(() -> { + List> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions(); + for (StateMachineAccess a : withAllRegions) { + a.resetStateMachine(new DefaultStateMachineContext(state, null, null, null)); + } + })) + .then(stateMachine.startReactively()) + .thenMany(stateMachine.sendEvent(Mono.just(event))) + .then(); + } + + /** + * Adds the persist state change listener. + * + * @param listener the listener + */ + public void addPersistStateChangeListener(GenericPersistStateChangeListener listener) { + listeners.register(listener); + } + + /** + * The listener interface for receiving persistStateChange events. + * The class that is interested in processing a persistStateChange + * event implements this interface, and the object created + * with that class is registered with a component using the + * component's addPersistStateChangeListener method. When + * the persistStateChange event occurs, that object's appropriate + * method is invoked. + */ + public interface GenericPersistStateChangeListener { + + /** + * Called when state needs to be persisted. + * + * @param state the state + * @param message the message + * @param transition the transition + * @param stateMachine the state machine + */ + void onPersist(State state, Message message, Transition transition, + StateMachine stateMachine); + } + + private class PersistingStateChangeInterceptor extends StateMachineInterceptorAdapter { + + @Override + public void preStateChange(State state, Message message, + Transition transition, StateMachine stateMachine, + StateMachine rootStateMachine) { + listeners.onPersist(state, message, transition, stateMachine); + } + } + + private class CompositePersistStateChangeListener extends AbstractCompositeListener> implements + GenericPersistStateChangeListener { + + @Override + public void onPersist(State state, Message message, + Transition transition, StateMachine stateMachine) { + for (Iterator> iterator = getListeners().reverse(); iterator.hasNext(); ) { + GenericPersistStateChangeListener listener = iterator.next(); + listener.onPersist(state, message, transition, stateMachine); + } + } + } + +} diff --git a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/FactoryPersistStateMachineHandler.java b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/FactoryPersistStateMachineHandler.java new file mode 100644 index 00000000..29066e07 --- /dev/null +++ b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/FactoryPersistStateMachineHandler.java @@ -0,0 +1,50 @@ +package org.springframework.statemachine.recipes.persist; + +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineException; +import org.springframework.statemachine.config.StateMachineBuilder; +import org.springframework.statemachine.config.StateMachineFactory; +import org.springframework.util.Assert; + +/** + * {@code FactoryPersistStateMachineHandler} is a recipe can be used to + * handle a state change of an arbitrary entity in a persistent storage. + *
+ * This implementation accepts {@link StateMachineFactory} + * or {@link StateMachineBuilder.Builder} to provide thread safe feature + * without sharing same state machine in concurrent environment. + * New state machine will be created when handling method is called. + * + * @author Ng Zouyiu + */ +public class FactoryPersistStateMachineHandler extends AbstractPersistStateMachineHandler { + + protected final StateMachineFactory factory; + protected final StateMachineBuilder.Builder builder; + + public FactoryPersistStateMachineHandler(StateMachineBuilder.Builder builder) { + Assert.notNull(builder, "State machine builder must be set"); + this.builder = builder; + factory = null; + } + + public FactoryPersistStateMachineHandler(StateMachineFactory factory) { + Assert.notNull(factory, "State machine factory must be set"); + this.factory = factory; + builder = null; + } + + @Override + protected StateMachine getInitStateMachine() { + StateMachine stateMachine; + if (factory != null) { + stateMachine = factory.getStateMachine(); + } else if (builder != null) { + stateMachine = builder.build(); + } else { + throw new StateMachineException("Factory or builder must be set to build state machine for handler"); + } + initStateMachine(stateMachine); + return stateMachine; + } +} diff --git a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/GenericPersistStateMachineHandler.java b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/GenericPersistStateMachineHandler.java new file mode 100644 index 00000000..253017d9 --- /dev/null +++ b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/GenericPersistStateMachineHandler.java @@ -0,0 +1,53 @@ +/* + * Copyright 2015-2019 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 + * + * https://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.recipes.persist; + +import org.springframework.statemachine.StateMachine; +import org.springframework.util.Assert; + +/** + * {@code GenericPersistStateMachineHandler} is a recipe which can be used to + * handle a state change of an arbitrary entity in a persistent storage. + *
+ * For concurrent usage, please consider using {@link FactoryPersistStateMachineHandler} + * to provide thread safe feature instead. + * + * @author Janne Valkealahti + */ +public class GenericPersistStateMachineHandler extends AbstractPersistStateMachineHandler { + + protected final StateMachine stateMachine; + + /** + * Instantiates a new persist state machine handler. + * + * @param stateMachine the state machine + */ + public GenericPersistStateMachineHandler(StateMachine stateMachine) { + Assert.notNull(stateMachine, "State machine must be set"); + this.stateMachine = stateMachine; + } + + @Override + protected void onInit() throws Exception { + initStateMachine(stateMachine); + } + + @Override + protected StateMachine getInitStateMachine() { + return stateMachine; + } +} diff --git a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java index 6b5179a9..0d6564f1 100644 --- a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java +++ b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java @@ -15,143 +15,24 @@ */ package org.springframework.statemachine.recipes.persist; -import java.util.Iterator; -import java.util.List; - -import org.springframework.messaging.Message; import org.springframework.statemachine.StateMachine; -import org.springframework.statemachine.access.StateMachineAccess; -import org.springframework.statemachine.listener.AbstractCompositeListener; -import org.springframework.statemachine.state.State; -import org.springframework.statemachine.support.DefaultStateMachineContext; -import org.springframework.statemachine.support.LifecycleObjectSupport; -import org.springframework.statemachine.support.StateMachineInterceptorAdapter; -import org.springframework.statemachine.transition.Transition; -import org.springframework.util.Assert; - -import reactor.core.publisher.Mono; /** * {@code PersistStateMachineHandler} is a recipe which can be used to * handle a state change of an arbitrary entity in a persistent storage. + *
+ * For concurrent usage, please consider using {@link FactoryPersistStateMachineHandler} + * to provide thread safe feature instead. * * @author Janne Valkealahti - * */ -public class PersistStateMachineHandler extends LifecycleObjectSupport { +public class PersistStateMachineHandler extends GenericPersistStateMachineHandler { - private final StateMachine stateMachine; - private final PersistingStateChangeInterceptor interceptor = new PersistingStateChangeInterceptor(); - private final CompositePersistStateChangeListener listeners = new CompositePersistStateChangeListener(); + public PersistStateMachineHandler(StateMachine stateMachine) { + super(stateMachine); + } - /** - * Instantiates a new persist state machine handler. - * - * @param stateMachine the state machine - */ - public PersistStateMachineHandler(StateMachine stateMachine) { - Assert.notNull(stateMachine, "State machine must be set"); - this.stateMachine = stateMachine; - } - - @Override - protected void onInit() throws Exception { - stateMachine.getStateMachineAccessor().doWithAllRegions(function -> function.addStateMachineInterceptor(interceptor)); - } - - /** - * Handle event with entity. - * - * @param event the event - * @param state the state - * @return true if event was accepted - * @see #handleEventWithStateReactively(Message, String) - */ - @Deprecated - public boolean handleEventWithState(Message event, String state) { - stateMachine.stopReactively().block(); - List> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions(); - for (StateMachineAccess a : withAllRegions) { - a.resetStateMachine(new DefaultStateMachineContext(state, null, null, null)); - } - stateMachine.startReactively().block(); - return stateMachine.sendEvent(event); - } - - /** - * Handle event with entity reactively. - * - * @param event the event - * @param state the state - * @return mono for completion - */ - public Mono handleEventWithStateReactively(Message event, String state) { - // TODO: REACTOR add docs and revisit this function concept - return Mono.from(stateMachine.stopReactively()) - .then(Mono.fromRunnable(() -> { - List> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions(); - for (StateMachineAccess a : withAllRegions) { - a.resetStateMachine(new DefaultStateMachineContext(state, null, null, null)); - } - })) - .then(stateMachine.startReactively()) - .thenMany(stateMachine.sendEvent(Mono.just(event))) - .then(); - } - - /** - * Adds the persist state change listener. - * - * @param listener the listener - */ - public void addPersistStateChangeListener(PersistStateChangeListener listener) { - listeners.register(listener); - } - - /** - * The listener interface for receiving persistStateChange events. - * The class that is interested in processing a persistStateChange - * event implements this interface, and the object created - * with that class is registered with a component using the - * component's addPersistStateChangeListener method. When - * the persistStateChange event occurs, that object's appropriate - * method is invoked. - */ - public interface PersistStateChangeListener { - - /** - * Called when state needs to be persisted. - * - * @param state the state - * @param message the message - * @param transition the transition - * @param stateMachine the state machine - */ - void onPersist(State state, Message message, Transition transition, - StateMachine stateMachine); - } - - private class PersistingStateChangeInterceptor extends StateMachineInterceptorAdapter { - - @Override - public void preStateChange(State state, Message message, - Transition transition, StateMachine stateMachine, - StateMachine rootStateMachine) { - listeners.onPersist(state, message, transition, stateMachine); - } - } - - private class CompositePersistStateChangeListener extends AbstractCompositeListener implements - PersistStateChangeListener { - - @Override - public void onPersist(State state, Message message, - Transition transition, StateMachine stateMachine) { - for (Iterator iterator = getListeners().reverse(); iterator.hasNext();) { - PersistStateChangeListener listener = iterator.next(); - listener.onPersist(state, message, transition, stateMachine); - } - } - } + public interface PersistStateChangeListener extends GenericPersistStateChangeListener { + } } diff --git a/spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/PersistStateMachineHandlerTests.java b/spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/PersistStateMachineHandlerTests.java index aac4b2ff..308d4a57 100644 --- a/spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/PersistStateMachineHandlerTests.java +++ b/spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/PersistStateMachineHandlerTests.java @@ -23,12 +23,14 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; import org.junit.jupiter.api.Test; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.config.StateMachineBuilder; +import org.springframework.statemachine.recipes.persist.FactoryPersistStateMachineHandler; import org.springframework.statemachine.recipes.persist.PersistStateMachineHandler; import org.springframework.statemachine.recipes.persist.PersistStateMachineHandler.PersistStateChangeListener; import org.springframework.statemachine.state.State; @@ -114,6 +116,47 @@ public class PersistStateMachineHandlerTests { assertThat(stateMachine.getState().getIds(), containsInAnyOrder("S2")); } + @Test + public void testFactoryPersistStateMachineHandler() throws Exception { + StateMachineBuilder.Builder builder = testStateMachineBuilder(); + + FactoryPersistStateMachineHandler handler = new FactoryPersistStateMachineHandler(builder); + + TestPersistStateChangeListener listener = new TestPersistStateChangeListener(); + handler.addPersistStateChangeListener(listener); + + Message event = MessageBuilder.withPayload("E2").build(); + handler.handleEventWithStateReactively(event, "S1").subscribe(); + + assertThat(listener.latch.await(1, TimeUnit.SECONDS), is(true)); + } + + @Test + public void testConcurrentFactoryPersistStateMachineHandler() throws Exception { + StateMachineBuilder.Builder builder = testStateMachineBuilder(); + + FactoryPersistStateMachineHandler handler = new FactoryPersistStateMachineHandler(builder); + + ArrayList listeners = new ArrayList<>(); + for (int i = 0; i < 8; i++) { + listeners.add(new TestPersistStateChangeListener()); + } + + for (TestPersistStateChangeListener listener : listeners) { + new Thread(() -> { + handler.addPersistStateChangeListener(listener); + + Message event = MessageBuilder.withPayload("E2").build(); + handler.handleEventWithStateReactively(event, "S1").subscribe(); + }).start(); + } + + for (TestPersistStateChangeListener listener : listeners) { + assertThat(listener.latch.await(1, TimeUnit.SECONDS), is(true)); + } + } + + private static class TestPersistStateChangeListener implements PersistStateChangeListener { CountDownLatch latch = new CountDownLatch(1); @@ -128,8 +171,7 @@ public class PersistStateMachineHandlerTests { } - private static StateMachine buildTestStateMachine() - throws Exception { + private static StateMachineBuilder.Builder testStateMachineBuilder() throws Exception { StateMachineBuilder.Builder builder = StateMachineBuilder.builder(); builder.configureConfiguration() @@ -149,11 +191,14 @@ public class PersistStateMachineHandlerTests { .withExternal() .source("S1").target("S2").event("E2"); - return builder.build(); + return builder; } - private static StateMachine buildTestStateMachine2() - throws Exception { + private static StateMachine buildTestStateMachine() throws Exception { + return testStateMachineBuilder().build(); + } + + private static StateMachineBuilder.Builder testStateMachineBuilder2() throws Exception { StateMachineBuilder.Builder builder = StateMachineBuilder.builder(); builder.configureConfiguration() @@ -175,6 +220,10 @@ public class PersistStateMachineHandlerTests { .source("S1") .last("S2"); - return builder.build(); + return builder; + } + + private static StateMachine buildTestStateMachine2() throws Exception { + return testStateMachineBuilder2().build(); } }