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

Move anonymous transitions to reactive chain

Browse Source 
- This commit changes a way how triggerless transitions are executed by
  going via new doOnComplete method in StateListener which returns
  Mono<Void>. This used to be a simple fire and forget subscribe via listener
  and now fully handled via reactive chain when state is complete. Rest of
  a changes are to tweak state actions to run parallel to be able to cancel
  those and then follow and track when triggerless transitions need to be
  executed.
- AbstractState still have some work to do for disposing things around
  submachines which currently seem to break thins if handleStateDoOnComplete
  is disposed when submachine state is exited. We'll leave this to get
  fixed later.
- Add tag handling for junit5 which can be set via gradle build properties
  statemachineIncludeTags and statemachineExcludeTags.
- Add BlockHound to build which can be activated via gradle build
  property statemachineBlockHound.
- Add org.awaitility:awaitility to various test deps.
- Mostly relates to #734
This commit is contained in:
Janne Valkealahti
2019-05-25 13:53:31 +01:00
parent d0c692c2e8
commit 5562bfdd12
13 changed files with 396 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
build.gradle
View File

@@ -15,6 +15,7 @@ buildscript {
curatorVersion = '2.11.1'
docResourcesVersion = '0.1.1.RELEASE'
awaitilityVersion = '3.1.6'
reactorBlockHoundVersion = '1.0.0.M3'
}
repositories {
maven { url 'https://repo.springsource.org/libs-release'}
@@ -103,6 +104,7 @@ configure(allprojects) {
dependency "org.apache.curator:curator-recipes:$curatorVersion"
dependency "org.apache.curator:curator-test:$curatorVersion"
dependency "org.awaitility:awaitility:$awaitilityVersion"
dependency "io.projectreactor.tools:blockhound-junit-platform:$reactorBlockHoundVersion"
}
}
@@ -111,7 +113,14 @@ configure(allprojects) {
}
test {
useJUnitPlatform()
useJUnitPlatform {
if (project.hasProperty('statemachineIncludeTags') && statemachineIncludeTags.size() > 0) {
includeTags = statemachineIncludeTags.split(',')
}
if (project.hasProperty('statemachineExcludeTags') && statemachineExcludeTags.size() > 0) {
excludeTags = statemachineExcludeTags.split(',')
}
}
exclude '**/*IntegrationTests.*'
}
}
@@ -123,6 +132,10 @@ configure(subprojects) { subproject ->
testCompile("org.junit.jupiter:junit-jupiter-api")
testRuntime("org.junit.jupiter:junit-jupiter-engine")
testRuntime("org.junit.vintage:junit-vintage-engine")
if (project.hasProperty('statemachineBlockHound') && statemachineBlockHound.toBoolean()) {
testRuntime("org.junit.platform:junit-platform-launcher")
testRuntime("io.projectreactor.tools:blockhound-junit-platform")
}
}
jar {
@@ -375,6 +388,7 @@ project('spring-statemachine-uml') {
testCompile 'org.hamcrest:hamcrest-core'
testCompile 'org.hamcrest:hamcrest-library'
testCompile 'junit:junit'
testCompile 'org.awaitility:awaitility'
testRuntime 'org.apache.logging.log4j:log4j-core'
}
}

278
spring-statemachine-core/src/main/java/org/springframework/statemachine/state/AbstractState.java
View File

@@ -15,21 +15,21 @@
*/
package org.springframework.statemachine.state;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Subscription;
import org.springframework.messaging.Message;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.statemachine.StateContext;
import org.springframework.statemachine.StateContext.Stage;
import org.springframework.statemachine.StateMachine;
@@ -44,8 +44,10 @@ import org.springframework.statemachine.support.LifecycleObjectSupport;
import org.springframework.statemachine.support.StateMachineUtils;
import org.springframework.statemachine.trigger.Trigger;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
/**
* Base implementation of a {@link State}.
@@ -64,15 +66,16 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
private final Collection<Function<StateContext<S, E>, Mono<Void>>> entryActions;
private final Collection<Function<StateContext<S, E>, Mono<Void>>> exitActions;
private final Collection<Function<StateContext<S, E>, Mono<Void>>> stateActions;
private final List<ScheduledAction> scheduledActions = new ArrayList<>();
private final Collection<Region<S, E>> regions = new ArrayList<Region<S, E>>();
private final StateMachine<S, E> submachine;
private List<Trigger<S, E>> triggers = new ArrayList<Trigger<S, E>>();
private final CompositeStateListener<S, E> stateListener = new CompositeStateListener<S, E>();
private final List<ScheduledAction> scheduledActions = new ArrayList<>();
private CompositeActionListener<S, E> actionListener;
private final List<StateMachineListener<S, E>> completionListeners = new CopyOnWriteArrayList<>();
private StateDoActionPolicy stateDoActionPolicy;
private Long stateDoActionPolicyTimeout;
private final Queue<Disposable> disposables = new ConcurrentLinkedDeque<>();
/**
* Instantiates a new abstract state.
@@ -219,7 +222,7 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
@Override
public Mono<Void> exit(StateContext<S, E> context) {
return Mono.defer(() -> {
return Mono.<Void>defer(() -> {
if (submachine != null) {
for (StateMachineListener<S, E> l : completionListeners) {
submachine.removeStateListener(l);
@@ -231,61 +234,80 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
}
}
}
return Mono.empty();
})
.then(Mono.<Void>fromRunnable(() -> {
completionListeners.clear();
cancelStateActions();
}))
.then(cancelStateActions())
.then(Mono.<Void>fromRunnable(() -> {
stateListener.onExit(context);
disarmTriggers();
return Mono.empty();
});
}))
.doFinally(signal -> disposeDisposables());
}
@Override
public Mono<Void> entry(StateContext<S, E> context) {
return Mono.defer(() -> {
if (submachine != null) {
final StateMachineListener<S, E> l = new StateMachineListenerAdapter<S, E>() {
Disposable disposable = Mono.just(submachine)
.flatMap(submachine -> {
return Mono.<Void>create(sink -> {
final StateMachineListener<S, E> l = new StateMachineListenerAdapter<S, E>() {
@Override
public void stateContext(StateContext<S, E> stateContext) {
if (stateContext.getStage() == Stage.STATEMACHINE_STOP) {
if (stateContext.getStateMachine() == submachine && submachine.isComplete()) {
completionListeners.remove(this);
submachine.removeStateListener(this);
if (completionListeners.isEmpty()) {
notifyStateOnComplete(stateContext);
}
}
}
}
};
submachine.addStateListener(l);
} else if (!regions.isEmpty()) {
for (final Region<S, E> region : regions) {
final StateMachineListener<S, E> l = new StateMachineListenerAdapter<S, E>() {
@Override
public void stateContext(StateContext<S, E> stateContext) {
if (stateContext.getStage() == Stage.STATEMACHINE_STOP) {
if (stateContext.getStateMachine() == region && region.isComplete()) {
completionListeners.remove(this);
region.removeStateListener(this);
if (completionListeners.isEmpty()) {
notifyStateOnComplete(stateContext);
@Override
public void stateContext(StateContext<S, E> stateContext) {
if (stateContext.getStage() == Stage.STATEMACHINE_STOP) {
if (stateContext.getStateMachine() == submachine && submachine.isComplete()) {
completionListeners.remove(this);
submachine.removeStateListener(this);
if (completionListeners.isEmpty()) {
sink.success();
}
}
}
}
}
}
};
completionListeners.add(l);
region.addStateListener(l);
}
}
};
submachine.addStateListener(l);
});
})
// TODO: REACTOR this is causing cancel which breaks some things
// .then(handleStateDoOnComplete(context))
.then(Mono.fromRunnable(() -> notifyStateOnComplete(context)))
.subscribe();
disposables.add(disposable);
} else if (!regions.isEmpty()) {
// TODO: REACTOR we should handle disposable
Flux.fromIterable(regions)
.flatMap(region -> {
return Mono.<Void>create(sink -> {
final StateMachineListener<S, E> l = new StateMachineListenerAdapter<S, E>() {
@Override
public void stateContext(StateContext<S, E> stateContext) {
if (stateContext.getStage() == Stage.STATEMACHINE_STOP) {
if (stateContext.getStateMachine() == region && region.isComplete()) {
completionListeners.remove(this);
region.removeStateListener(this);
sink.success();
}
}
}
};
completionListeners.add(l);
region.addStateListener(l);
});
})
.then(handleStateDoOnComplete(context))
.then(Mono.fromRunnable(() -> notifyStateOnComplete(context)))
.subscribe();
}
stateListener.onEntry(context);
armTriggers();
scheduleStateActions(context);
return Mono.empty();
});
})
.then(scheduleStateActions(context));
}
@Override
@@ -449,59 +471,80 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
}
}
/**
* Cancel existing state actions and clear list.
*/
protected void cancelStateActions() {
if (log.isDebugEnabled()) {
log.debug("Handling finish of state actions, scheduledActions size is " + scheduledActions.size());
private void disposeDisposables() {
Disposable disposable;
while ((disposable = disposables.poll()) != null) {
disposable.dispose();
}
for (ScheduledAction task : scheduledActions) {
if (task.timeout != null) {
if (log.isDebugEnabled()) {
log.debug("Timeouting scheduled state do action " + task);
}
try {
task.future.get(task.timeout, TimeUnit.MILLISECONDS);
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug("Cancelling scheduled state do action after timeout " + task);
}
task.future.cancel(true);
}
} else {
if (log.isDebugEnabled()) {
log.debug("Cancelling scheduled state do action immediately " + task);
}
task.future.cancel(true);
}
}
scheduledActions.clear();
}
/**
* Schedule state actions and store futures into list to
* be cancelled.
*
* @param context the context
*/
protected void scheduleStateActions(StateContext<S, E> context) {
AtomicInteger completionCount = null;
if (isSimple()) {
completionCount = new AtomicInteger(stateActions.size());
}
for (Function<StateContext<S, E>, Mono<Void>> action : stateActions) {
ScheduledFuture<?> future = scheduleAction(action, context, completionCount);
if (log.isDebugEnabled()) {
log.debug("Scheduling state do action " + action + " with future " + future);
private Mono<Void> scheduleStateActions(StateContext<S, E> context) {
return Mono.defer(() -> {
final AtomicInteger completionCount = new AtomicInteger(stateActions.size());
Long timeout = resolveDoActionTimeout(context);
return Flux.fromIterable(stateActions)
.map(stateAction -> executeAction(stateAction, context))
.map(function -> {
return function
.subscribeOn(Schedulers.parallel())
.doOnSubscribe(subscription -> {
if (log.isDebugEnabled()) {
log.debug("Adding new scheduled action with subscription=" + subscription);
}
scheduledActions.add(new ScheduledAction(subscription, timeout, System.currentTimeMillis()));
})
.then(handleCompleteOrEmpty1(context, completionCount))
.subscribe();
})
.then(handleCompleteOrEmpty2(context, completionCount))
;
});
}
private Mono<Void> handleCompleteOrEmpty1(StateContext<S, E> context, AtomicInteger completionCount) {
return Mono.defer(() -> {
if (completionCount.decrementAndGet() <= 0 && stateActions.size() > 0) {
return handleStateDoOnComplete(context)
.then(Mono.fromRunnable(() -> notifyStateOnComplete(context)));
} else {
return Mono.empty();
}
if (future != null) {
scheduledActions.add(new ScheduledAction(future, resolveDoActionTimeout(context)));
});
}
private Mono<Void> handleCompleteOrEmpty2(StateContext<S, E> context, AtomicInteger completionCount) {
return Mono.defer(() -> {
if (isSimple() && stateActions.size() == 0) {
return handleStateDoOnComplete(context)
.then(Mono.fromRunnable(() -> notifyStateOnComplete(context)));
} else {
return Mono.empty();
}
}
if (isSimple() && stateActions.size() == 0) {
notifyStateOnComplete(context);
}
});
}
private Mono<Void> cancelStateActions() {
return Flux.fromIterable(scheduledActions)
// state action tells us how long it needs for timeout, delay
.flatMap(stateAction -> {
// check delay and prevent unnecessary thread switch with Mono.delay()
if (stateAction.getNeededDelayNow().toMillis() > 0) {
return Mono.delay(stateAction.getNeededDelayNow()).thenReturn(stateAction);
} else {
return Mono.just(stateAction);
}
})
// then dispose which i.e. should interrupt blocking threads or cancel reactive code
.doOnNext(stateAction -> {
if (stateAction.subscription != null) {
log.debug("About to dispose subscription " + stateAction.subscription);
stateAction.subscription.cancel();
}
})
// we're done, clear state scheduled state actions
.thenEmpty(Mono.fromRunnable(() -> {
scheduledActions.clear();
}));
}
/**
@@ -528,33 +571,8 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
});
}
/**
* Schedule action and return future which can be used to cancel it.
*
* @param action the action
* @param context the context
* @param completionCount the completion count tracker
* @return the scheduled future
*/
protected ScheduledFuture<?> scheduleAction(final Function<StateContext<S, E>, Mono<Void>> action, final StateContext<S, E> context,
final AtomicInteger completionCount) {
TaskScheduler taskScheduler = getTaskScheduler();
if (taskScheduler == null) {
log.error("Unable to schedule action as taskSchedule is not set, action=[" + action + "]");
return null;
}
ScheduledFuture<?> future = taskScheduler.schedule(new Runnable() {
@Override
public void run() {
// TODO: REACTOR subscribe is probably wrong!
executeAction(action, context).subscribe();
if (completionCount != null && completionCount.decrementAndGet() <= 0) {
notifyStateOnComplete(context);
}
}
}, new Date());
return future;
protected Mono<Void> handleStateDoOnComplete(StateContext<S, E> context) {
return stateListener.doOnComplete(context);
}
protected void notifyStateOnComplete(StateContext<S, E> context) {
@@ -573,16 +591,24 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
}
private static class ScheduledAction {
ScheduledFuture<?> future;
Subscription subscription;
Long timeout;
Long subscribeTime;
public ScheduledAction(ScheduledFuture<?> future, Long timeout) {
this.future = future;
ScheduledAction(Subscription subscription, Long timeout, Long subscribeTime) {
this.subscription = subscription;
this.timeout = timeout;
this.subscribeTime = subscribeTime;
}
@Override
public String toString() {
return "ScheduledTask [future=" + future + ", timeout=" + timeout + "]";
Duration getNeededDelayNow() {
long delay = 0;
if (subscribeTime != null && timeout != null) {
long now = System.currentTimeMillis();
long tocancel = subscribeTime + timeout;
delay = now > tocancel ? 0 : tocancel - now;
}
return Duration.ofMillis(delay);
}
}

19
spring-statemachine-core/src/main/java/org/springframework/statemachine/state/CompositeStateListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-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.
@@ -16,10 +16,15 @@
package org.springframework.statemachine.state;
import java.util.Iterator;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import org.springframework.statemachine.StateContext;
import org.springframework.statemachine.listener.AbstractCompositeListener;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Composite state listener.
*
@@ -51,4 +56,16 @@ public class CompositeStateListener<S, E> extends AbstractCompositeListener<Stat
iterator.next().onComplete(context);
}
}
@Override
public Mono<Void> doOnComplete(StateContext<S, E> context) {
return Mono.defer(() -> {
Iterator<StateListener<S, E>> iterator = getListeners().reverse();
Iterable<StateListener<S, E>> iterable = () -> iterator;
Stream<StateListener<S, E>> stream = StreamSupport.stream(iterable.spliterator(), false);
return Flux.fromStream(stream)
.flatMap(listener -> listener.doOnComplete(context))
.then();
});
}
}

6
spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ObjectState.java
View File

@@ -146,6 +146,9 @@ public class ObjectState<S, E> extends AbstractSimpleState<S, E> {
public Mono<Void> exit(StateContext<S, E> context) {
Mono<Void> actions = Flux.fromIterable(getExitActions())
.flatMap(a -> executeAction(a, context))
.onErrorContinue((t, u) -> {
// TODO: REACTOR allow continue and fix with error handling overhaul
})
.then();
return super.exit(context).and(actions);
}
@@ -154,6 +157,9 @@ public class ObjectState<S, E> extends AbstractSimpleState<S, E> {
public Mono<Void> entry(StateContext<S, E> context) {
Mono<Void> actions = Flux.fromIterable(getEntryActions())
.flatMap(a -> executeAction(a, context))
.onErrorContinue((t, u) -> {
// TODO: REACTOR allow continue and fix with error handling overhaul
})
.then();
return actions.and(super.entry(context));
}

12
spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-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.
@@ -17,6 +17,8 @@ package org.springframework.statemachine.state;
import org.springframework.statemachine.StateContext;
import reactor.core.publisher.Mono;
/**
* {@code StateListener} for various state events.
*
@@ -47,4 +49,12 @@ public interface StateListener<S, E> {
* @param context the state context
*/
void onComplete(StateContext<S, E> context);
/**
* Called when {@link State} want to notify of its completion.
*
* @param context the state context
* @return mono for completion
*/
Mono<Void> doOnComplete(StateContext<S, E> context);
}

9
spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListenerAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018 the original author or authors.
* Copyright 2018-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.
@@ -17,6 +17,8 @@ package org.springframework.statemachine.state;
import org.springframework.statemachine.StateContext;
import reactor.core.publisher.Mono;
/**
* Adapter implementation of {@link StateListener} implementing all
* methods which extended implementation can override.
@@ -39,4 +41,9 @@ public class StateListenerAdapter<S, E> implements StateListener<S, E> {
@Override
public void onComplete(StateContext<S, E> context) {
}
@Override
public Mono<Void> doOnComplete(StateContext<S, E> context) {
return Mono.empty();
}
}

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

@@ -277,11 +277,14 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
for (final State<S, E> state : states) {
state.addStateListener(new StateListenerAdapter<S, E>() {
@Override
public void onComplete(StateContext<S, E> context) {
log.debug("State onComplete: state=[" + state + "] context=[" + context + "]");
((AbstractStateMachine<S, E>)getRelayStateMachine()).executeTriggerlessTransitions(AbstractStateMachine.this, context, state).subscribe();
};
public Mono<Void> doOnComplete(StateContext<S, E> context) {
if (log.isDebugEnabled()) {
log.debug("State onComplete: state=[" + state + "] context=[" + context + "]");
}
return ((AbstractStateMachine<S, E>)getRelayStateMachine()).executeTriggerlessTransitions(AbstractStateMachine.this, context, state);
}
});
if (state.isSubmachineState()) {

5
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java
View File

@@ -56,7 +56,7 @@ import reactor.core.publisher.Mono;
/**
* Default reactive implementation of a {@link StateMachineExecutor}.
*
*
* @author Janne Valkealahti
*
* @param <S> the type of state
@@ -168,6 +168,9 @@ public class ReactiveStateMachineExecutor<S, E> extends LifecycleObjectSupport i
@Override
public Mono<Void> executeTriggerlessTransitions(StateContext<S, E> context, State<S, E> state) {
if (stateMachine.getState() != null) {
if (log.isDebugEnabled()) {
log.debug("About to handleTriggerlessTransitions");
}
return handleTriggerlessTransitions(context, state);
}
return Mono.empty();

6
spring-statemachine-core/src/test/java/org/springframework/statemachine/action/StateDoActivityActionTests.java
View File

@@ -68,8 +68,8 @@ public class StateDoActivityActionTests extends AbstractStateMachineTests {
assertThat(testActionS1.onExecuteStartLatch.await(2, TimeUnit.SECONDS), is(true));
doSendEventAndConsumeAll(machine, TestEvents.E1);
assertThat(testActionS1.interruptedLatch.await(2, TimeUnit.SECONDS), is(true));
assertThat(testActionS1.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true));
assertThat(testActionS1.interruptedLatch.await(6, TimeUnit.SECONDS), is(true));
assertThat(testActionS1.onExecuteLatch.await(6, TimeUnit.SECONDS), is(true));
assertThat(testActionS2.onExecuteStartLatch.await(2, TimeUnit.SECONDS), is(true));
doSendEventAndConsumeAll(machine, TestEvents.E2);
@@ -91,7 +91,9 @@ public class StateDoActivityActionTests extends AbstractStateMachineTests {
doSendEventAndConsumeAll(machine, TestEvents.E3);
assertThat(testActionS1I.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true));
assertThat(testActionS1.interruptedLatch.await(2, TimeUnit.SECONDS), is(false));
doSendEventAndConsumeAll(machine, TestEvents.E1);
assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S2));
doSendEventAndConsumeAll(machine, TestEvents.E4);
assertThat(testActionS2I.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true));

1
spring-statemachine-core/src/test/java/org/springframework/statemachine/monitor/StateMachineMonitorTests.java
View File

@@ -195,7 +195,6 @@ public class StateMachineMonitorTests extends AbstractStateMachineTests {
@Override
public void action(StateMachine<String, String> stateMachine,
Function<StateContext<String, String>, Mono<Void>> action, long duration) {
System.out.println("XXX HI");
actions.put(action, new Actions(action, duration));
latch.countDown();
}

4
spring-statemachine-core/src/test/java/org/springframework/statemachine/state/CompletionEventTests.java
View File

@@ -15,6 +15,8 @@
*/
package org.springframework.statemachine.state;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import static org.springframework.statemachine.TestUtils.doSendEventAndConsumeAll;
@@ -57,7 +59,7 @@ public class CompletionEventTests extends AbstractStateMachineTests {
assertThat(testAction2.latch.await(2, TimeUnit.SECONDS), is(true));
assertThat(testAction2.count, is(1));
Thread.sleep(1000);
await().until(() -> machine.getState().getIds(), containsInAnyOrder("S3"));
assertThat(machine.getState().getId(), is("S3"));
}

167
spring-statemachine-core/src/test/java/org/springframework/statemachine/state/ObjectStateTests.java Normal file
View File

@@ -0,0 +1,167 @@
/*
* Copyright 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.state;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.CoreMatchers.is;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
import org.springframework.statemachine.StateContext;
import org.springframework.statemachine.action.ReactiveAction;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
public class ObjectStateTests {
@Test
public void testEntrySingleAction() {
TestAction action = new TestAction();
ObjectState<String, String> state = new ObjectState<>("TEST", null, Arrays.asList(action), null, null, null,
null, null);
StepVerifier.create(state.entry(null))
.expectComplete()
.verify();
assertThat(action.count).hasValue(1);
}
@Test
public void testEntryMultiActions() {
TestAction action1 = new TestAction();
TestAction action2 = new TestAction();
ObjectState<String, String> state = new ObjectState<>("TEST", null, Arrays.asList(action1, action2), null, null,
null, null, null);
StepVerifier.create(state.entry(null))
.expectComplete()
.verify();
assertThat(action1.count).hasValue(1);
assertThat(action2.count).hasValue(1);
}
@Test
public void testExitSingle() {
TestAction action = new TestAction();
ObjectState<String, String> state = new ObjectState<>("TEST", null, null, Arrays.asList(action), null, null,
null, null);
StepVerifier.create(state.exit(null))
.expectComplete()
.verify();
assertThat(action.count).hasValue(1);
}
@Test
public void testExitMultiActions() {
TestAction action1 = new TestAction();
TestAction action2 = new TestAction();
ObjectState<String, String> state = new ObjectState<>("TEST", null, null, Arrays.asList(action1, action2), null,
null, null, null);
StepVerifier.create(state.exit(null))
.expectComplete()
.verify();
assertThat(action1.count).hasValue(1);
assertThat(action2.count).hasValue(1);
}
@Test
public void testStateAction() {
TestAction action = new TestAction();
ObjectState<String, String> state = new ObjectState<>("TEST", null, null, null, Arrays.asList(action), null,
null, null);
StepVerifier.create(state.entry(null))
.expectComplete()
.verify();
await().untilAtomic(action.count, is(1));
}
@Test
public void testEntrySingleActionBlocks() {
TestBlockingAction action = new TestBlockingAction();
ObjectState<String, String> state = new ObjectState<>("TEST", null, Arrays.asList(action), null, null, null,
null, null);
StepVerifier.create(state.entry(null))
.expectComplete()
.verify();
assertThat(action.countBefore).hasValue(1);
assertThat(action.countInterrupt).hasValue(0);
assertThat(action.countAfter).hasValue(1);
}
@Test
public void testStateActionBlocks() {
TestBlockingAction action = new TestBlockingAction();
ObjectState<String, String> state = new ObjectState<>("TEST", null, null, null, Arrays.asList(action), null,
null, null);
StepVerifier.create(state.entry(null))
.expectComplete()
.verify();
await().untilAtomic(action.countAfter, is(1));
assertThat(action.countBefore).hasValue(1);
assertThat(action.countInterrupt).hasValue(0);
}
@Test
public void testStateMultiActionBlocks() {
TestBlockingAction action1 = new TestBlockingAction();
TestBlockingAction action2 = new TestBlockingAction();
ObjectState<String, String> state = new ObjectState<>("TEST", null, null, null, Arrays.asList(action1, action2),
null, null, null);
StepVerifier.create(state.entry(null))
.expectComplete()
.verify();
await().untilAtomic(action1.countAfter, is(1));
assertThat(action1.countBefore).hasValue(1);
assertThat(action1.countInterrupt).hasValue(0);
await().untilAtomic(action2.countAfter, is(1));
assertThat(action2.countBefore).hasValue(1);
assertThat(action2.countInterrupt).hasValue(0);
}
private static class TestAction implements ReactiveAction<String, String> {
AtomicInteger count = new AtomicInteger();
@Override
public Mono<Void> apply(StateContext<String, String> context) {
return Mono.<Void>empty().doOnSuccess(d -> {
count.incrementAndGet();
});
}
}
private static class TestBlockingAction implements ReactiveAction<String, String> {
AtomicInteger countBefore = new AtomicInteger();
AtomicInteger countAfter = new AtomicInteger();
AtomicInteger countInterrupt = new AtomicInteger();
@Override
public Mono<Void> apply(StateContext<String, String> context) {
return Mono.fromRunnable(() -> {
countBefore.incrementAndGet();
try {
Thread.sleep(1000);
} catch (Exception e) {
countInterrupt.incrementAndGet();
}
countAfter.incrementAndGet();
});
}
}
}

2
spring-statemachine-samples/ordershipping/src/test/java/demo/ordershipping/StateMachineTests.java
View File

@@ -106,7 +106,7 @@ public class StateMachineTests {
.setHeader("payment", "1000").build())
.expectStates("ORDER_SHIPPED")
// .expectStateChanged(4)
.expectStateChanged(3)
.expectStateChanged(2)
.expectStateMachineStopped(3)
.and()
.build();