Add generic thread safe support to PersistStateMachineHandler
This commit is contained in:
xJoeWoo
committed by
Janne Valkealahti
Janne Valkealahti
parent
61c58cdee0
commit
34a28b57fa
@@ -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<S, E> extends LifecycleObjectSupport {
|
||||
|
||||
protected final PersistingStateChangeInterceptor interceptor = new PersistingStateChangeInterceptor();
|
||||
protected final CompositePersistStateChangeListener listeners = new CompositePersistStateChangeListener();
|
||||
|
||||
protected abstract StateMachine<S, E> getInitStateMachine();
|
||||
|
||||
protected void initStateMachine(StateMachine<S, E> stateMachine) {
|
||||
List<StateMachineAccess<S, E>> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions();
|
||||
for (StateMachineAccess<S, E> 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<E> event, S state) {
|
||||
StateMachine<S, E> stateMachine = getInitStateMachine();
|
||||
stateMachine.stopReactively().block();
|
||||
List<StateMachineAccess<S, E>> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions();
|
||||
for (StateMachineAccess<S, E> a : withAllRegions) {
|
||||
a.resetStateMachine(new DefaultStateMachineContext<S, E>(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<Void> handleEventWithStateReactively(Message<E> event, S state) {
|
||||
StateMachine<S, E> stateMachine = getInitStateMachine();
|
||||
// TODO: REACTOR add docs and revisit this function concept
|
||||
return Mono.from(stateMachine.stopReactively())
|
||||
.then(Mono.fromRunnable(() -> {
|
||||
List<StateMachineAccess<S, E>> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions();
|
||||
for (StateMachineAccess<S, E> a : withAllRegions) {
|
||||
a.resetStateMachine(new DefaultStateMachineContext<S, E>(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<S, E> 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 <code>addPersistStateChangeListener</code> method. When
|
||||
* the persistStateChange event occurs, that object's appropriate
|
||||
* method is invoked.
|
||||
*/
|
||||
public interface GenericPersistStateChangeListener<S, E> {
|
||||
|
||||
/**
|
||||
* 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<S, E> state, Message<E> message, Transition<S, E> transition,
|
||||
StateMachine<S, E> stateMachine);
|
||||
}
|
||||
|
||||
private class PersistingStateChangeInterceptor extends StateMachineInterceptorAdapter<S, E> {
|
||||
|
||||
@Override
|
||||
public void preStateChange(State<S, E> state, Message<E> message,
|
||||
Transition<S, E> transition, StateMachine<S, E> stateMachine,
|
||||
StateMachine<S, E> rootStateMachine) {
|
||||
listeners.onPersist(state, message, transition, stateMachine);
|
||||
}
|
||||
}
|
||||
|
||||
private class CompositePersistStateChangeListener extends AbstractCompositeListener<GenericPersistStateChangeListener<S, E>> implements
|
||||
GenericPersistStateChangeListener<S, E> {
|
||||
|
||||
@Override
|
||||
public void onPersist(State<S, E> state, Message<E> message,
|
||||
Transition<S, E> transition, StateMachine<S, E> stateMachine) {
|
||||
for (Iterator<GenericPersistStateChangeListener<S, E>> iterator = getListeners().reverse(); iterator.hasNext(); ) {
|
||||
GenericPersistStateChangeListener<S, E> listener = iterator.next();
|
||||
listener.onPersist(state, message, transition, stateMachine);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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.
|
||||
* <br>
|
||||
* 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<S, E> extends AbstractPersistStateMachineHandler<S, E> {
|
||||
|
||||
protected final StateMachineFactory<S, E> factory;
|
||||
protected final StateMachineBuilder.Builder<S, E> builder;
|
||||
|
||||
public FactoryPersistStateMachineHandler(StateMachineBuilder.Builder<S, E> builder) {
|
||||
Assert.notNull(builder, "State machine builder must be set");
|
||||
this.builder = builder;
|
||||
factory = null;
|
||||
}
|
||||
|
||||
public FactoryPersistStateMachineHandler(StateMachineFactory<S, E> factory) {
|
||||
Assert.notNull(factory, "State machine factory must be set");
|
||||
this.factory = factory;
|
||||
builder = null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected StateMachine<S, E> getInitStateMachine() {
|
||||
StateMachine<S, E> 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;
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
* <br>
|
||||
* For concurrent usage, please consider using {@link FactoryPersistStateMachineHandler}
|
||||
* to provide thread safe feature instead.
|
||||
*
|
||||
* @author Janne Valkealahti
|
||||
*/
|
||||
public class GenericPersistStateMachineHandler<S, E> extends AbstractPersistStateMachineHandler<S, E> {
|
||||
|
||||
protected final StateMachine<S, E> stateMachine;
|
||||
|
||||
/**
|
||||
* Instantiates a new persist state machine handler.
|
||||
*
|
||||
* @param stateMachine the state machine
|
||||
*/
|
||||
public GenericPersistStateMachineHandler(StateMachine<S, E> stateMachine) {
|
||||
Assert.notNull(stateMachine, "State machine must be set");
|
||||
this.stateMachine = stateMachine;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onInit() throws Exception {
|
||||
initStateMachine(stateMachine);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected StateMachine<S, E> getInitStateMachine() {
|
||||
return stateMachine;
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
* <br>
|
||||
* 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<String, String> {
|
||||
|
||||
private final StateMachine<String, String> stateMachine;
|
||||
private final PersistingStateChangeInterceptor interceptor = new PersistingStateChangeInterceptor();
|
||||
private final CompositePersistStateChangeListener listeners = new CompositePersistStateChangeListener();
|
||||
public PersistStateMachineHandler(StateMachine<String, String> stateMachine) {
|
||||
super(stateMachine);
|
||||
}
|
||||
|
||||
/**
|
||||
* Instantiates a new persist state machine handler.
|
||||
*
|
||||
* @param stateMachine the state machine
|
||||
*/
|
||||
public PersistStateMachineHandler(StateMachine<String, String> 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<String> event, String state) {
|
||||
stateMachine.stopReactively().block();
|
||||
List<StateMachineAccess<String, String>> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions();
|
||||
for (StateMachineAccess<String, String> a : withAllRegions) {
|
||||
a.resetStateMachine(new DefaultStateMachineContext<String, String>(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<Void> handleEventWithStateReactively(Message<String> event, String state) {
|
||||
// TODO: REACTOR add docs and revisit this function concept
|
||||
return Mono.from(stateMachine.stopReactively())
|
||||
.then(Mono.fromRunnable(() -> {
|
||||
List<StateMachineAccess<String, String>> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions();
|
||||
for (StateMachineAccess<String, String> a : withAllRegions) {
|
||||
a.resetStateMachine(new DefaultStateMachineContext<String, String>(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 <code>addPersistStateChangeListener</code> 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<String, String> state, Message<String> message, Transition<String, String> transition,
|
||||
StateMachine<String, String> stateMachine);
|
||||
}
|
||||
|
||||
private class PersistingStateChangeInterceptor extends StateMachineInterceptorAdapter<String, String> {
|
||||
|
||||
@Override
|
||||
public void preStateChange(State<String, String> state, Message<String> message,
|
||||
Transition<String, String> transition, StateMachine<String, String> stateMachine,
|
||||
StateMachine<String, String> rootStateMachine) {
|
||||
listeners.onPersist(state, message, transition, stateMachine);
|
||||
}
|
||||
}
|
||||
|
||||
private class CompositePersistStateChangeListener extends AbstractCompositeListener<PersistStateChangeListener> implements
|
||||
PersistStateChangeListener {
|
||||
|
||||
@Override
|
||||
public void onPersist(State<String, String> state, Message<String> message,
|
||||
Transition<String, String> transition, StateMachine<String, String> stateMachine) {
|
||||
for (Iterator<PersistStateChangeListener> iterator = getListeners().reverse(); iterator.hasNext();) {
|
||||
PersistStateChangeListener listener = iterator.next();
|
||||
listener.onPersist(state, message, transition, stateMachine);
|
||||
}
|
||||
}
|
||||
}
|
||||
public interface PersistStateChangeListener extends GenericPersistStateChangeListener<String, String> {
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<String, String> builder = testStateMachineBuilder();
|
||||
|
||||
FactoryPersistStateMachineHandler<String, String> handler = new FactoryPersistStateMachineHandler<String, String>(builder);
|
||||
|
||||
TestPersistStateChangeListener listener = new TestPersistStateChangeListener();
|
||||
handler.addPersistStateChangeListener(listener);
|
||||
|
||||
Message<String> 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<String, String> builder = testStateMachineBuilder();
|
||||
|
||||
FactoryPersistStateMachineHandler<String, String> handler = new FactoryPersistStateMachineHandler<String, String>(builder);
|
||||
|
||||
ArrayList<TestPersistStateChangeListener> listeners = new ArrayList<>();
|
||||
for (int i = 0; i < 8; i++) {
|
||||
listeners.add(new TestPersistStateChangeListener());
|
||||
}
|
||||
|
||||
for (TestPersistStateChangeListener listener : listeners) {
|
||||
new Thread(() -> {
|
||||
handler.addPersistStateChangeListener(listener);
|
||||
|
||||
Message<String> 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<String, String> buildTestStateMachine()
|
||||
throws Exception {
|
||||
private static StateMachineBuilder.Builder<String, String> testStateMachineBuilder() throws Exception {
|
||||
StateMachineBuilder.Builder<String, String> 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<String, String> buildTestStateMachine2()
|
||||
throws Exception {
|
||||
private static StateMachine<String, String> buildTestStateMachine() throws Exception {
|
||||
return testStateMachineBuilder().build();
|
||||
}
|
||||
|
||||
private static StateMachineBuilder.Builder<String, String> testStateMachineBuilder2() throws Exception {
|
||||
StateMachineBuilder.Builder<String, String> builder = StateMachineBuilder.builder();
|
||||
|
||||
builder.configureConfiguration()
|
||||
@@ -175,6 +220,10 @@ public class PersistStateMachineHandlerTests {
|
||||
.source("S1")
|
||||
.last("S2");
|
||||
|
||||
return builder.build();
|
||||
return builder;
|
||||
}
|
||||
|
||||
private static StateMachine<String, String> buildTestStateMachine2() throws Exception {
|
||||
return testStateMachineBuilder2().build();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user