From 5562bfdd12c0f9f8379f340819416f3ada522eef Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Sat, 25 May 2019 13:53:31 +0100 Subject: [PATCH] Move anonymous transitions to reactive chain - This commit changes a way how triggerless transitions are executed by going via new doOnComplete method in StateListener which returns Mono. 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 --- build.gradle | 16 +- .../statemachine/state/AbstractState.java | 278 ++++++++++-------- .../state/CompositeStateListener.java | 19 +- .../statemachine/state/ObjectState.java | 6 + .../statemachine/state/StateListener.java | 12 +- .../state/StateListenerAdapter.java | 9 +- .../support/AbstractStateMachine.java | 11 +- .../support/ReactiveStateMachineExecutor.java | 5 +- .../action/StateDoActivityActionTests.java | 6 +- .../monitor/StateMachineMonitorTests.java | 1 - .../state/CompletionEventTests.java | 4 +- .../statemachine/state/ObjectStateTests.java | 167 +++++++++++ .../demo/ordershipping/StateMachineTests.java | 2 +- 13 files changed, 396 insertions(+), 140 deletions(-) create mode 100644 spring-statemachine-core/src/test/java/org/springframework/statemachine/state/ObjectStateTests.java diff --git a/build.gradle b/build.gradle index 90038846..7896499f 100644 --- a/build.gradle +++ b/build.gradle @@ -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' } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/AbstractState.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/AbstractState.java index b9745948..f40ada5e 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/AbstractState.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/AbstractState.java @@ -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 extends LifecycleObjectSupport impleme private final Collection, Mono>> entryActions; private final Collection, Mono>> exitActions; private final Collection, Mono>> stateActions; + private final List scheduledActions = new ArrayList<>(); private final Collection> regions = new ArrayList>(); private final StateMachine submachine; private List> triggers = new ArrayList>(); private final CompositeStateListener stateListener = new CompositeStateListener(); - private final List scheduledActions = new ArrayList<>(); private CompositeActionListener actionListener; private final List> completionListeners = new CopyOnWriteArrayList<>(); private StateDoActionPolicy stateDoActionPolicy; private Long stateDoActionPolicyTimeout; + private final Queue disposables = new ConcurrentLinkedDeque<>(); /** * Instantiates a new abstract state. @@ -219,7 +222,7 @@ public abstract class AbstractState extends LifecycleObjectSupport impleme @Override public Mono exit(StateContext context) { - return Mono.defer(() -> { + return Mono.defer(() -> { if (submachine != null) { for (StateMachineListener l : completionListeners) { submachine.removeStateListener(l); @@ -231,61 +234,80 @@ public abstract class AbstractState extends LifecycleObjectSupport impleme } } } + return Mono.empty(); + }) + .then(Mono.fromRunnable(() -> { completionListeners.clear(); - cancelStateActions(); + })) + .then(cancelStateActions()) + .then(Mono.fromRunnable(() -> { stateListener.onExit(context); disarmTriggers(); - return Mono.empty(); - }); + })) + .doFinally(signal -> disposeDisposables()); } @Override public Mono entry(StateContext context) { return Mono.defer(() -> { if (submachine != null) { - final StateMachineListener l = new StateMachineListenerAdapter() { + Disposable disposable = Mono.just(submachine) + .flatMap(submachine -> { + return Mono.create(sink -> { + final StateMachineListener l = new StateMachineListenerAdapter() { - @Override - public void stateContext(StateContext 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 region : regions) { - final StateMachineListener l = new StateMachineListenerAdapter() { - - @Override - public void stateContext(StateContext 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 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.create(sink -> { + final StateMachineListener l = new StateMachineListenerAdapter() { + @Override + public void stateContext(StateContext 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 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 context) { - AtomicInteger completionCount = null; - if (isSimple()) { - completionCount = new AtomicInteger(stateActions.size()); - } - for (Function, Mono> action : stateActions) { - ScheduledFuture future = scheduleAction(action, context, completionCount); - if (log.isDebugEnabled()) { - log.debug("Scheduling state do action " + action + " with future " + future); + private Mono scheduleStateActions(StateContext 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 handleCompleteOrEmpty1(StateContext 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 handleCompleteOrEmpty2(StateContext 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 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 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, Mono> action, final StateContext 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 handleStateDoOnComplete(StateContext context) { + return stateListener.doOnComplete(context); } protected void notifyStateOnComplete(StateContext context) { @@ -573,16 +591,24 @@ public abstract class AbstractState 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); } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/CompositeStateListener.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/CompositeStateListener.java index ccf9cf6e..20f8be1a 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/CompositeStateListener.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/CompositeStateListener.java @@ -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 extends AbstractCompositeListener doOnComplete(StateContext context) { + return Mono.defer(() -> { + Iterator> iterator = getListeners().reverse(); + Iterable> iterable = () -> iterator; + Stream> stream = StreamSupport.stream(iterable.spliterator(), false); + return Flux.fromStream(stream) + .flatMap(listener -> listener.doOnComplete(context)) + .then(); + }); + } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ObjectState.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ObjectState.java index 51cfa205..3fe3e163 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ObjectState.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/ObjectState.java @@ -146,6 +146,9 @@ public class ObjectState extends AbstractSimpleState { public Mono exit(StateContext context) { Mono 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 extends AbstractSimpleState { public Mono entry(StateContext context) { Mono 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)); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListener.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListener.java index 7fb67ad3..d0e393b5 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListener.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListener.java @@ -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 { * @param context the state context */ void onComplete(StateContext context); + + /** + * Called when {@link State} want to notify of its completion. + * + * @param context the state context + * @return mono for completion + */ + Mono doOnComplete(StateContext context); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListenerAdapter.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListenerAdapter.java index 2a6e1472..640226ec 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListenerAdapter.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateListenerAdapter.java @@ -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 implements StateListener { @Override public void onComplete(StateContext context) { } + + @Override + public Mono doOnComplete(StateContext context) { + return Mono.empty(); + } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java index 8a13c065..262fdd96 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java @@ -277,11 +277,14 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo for (final State state : states) { state.addStateListener(new StateListenerAdapter() { + @Override - public void onComplete(StateContext context) { - log.debug("State onComplete: state=[" + state + "] context=[" + context + "]"); - ((AbstractStateMachine)getRelayStateMachine()).executeTriggerlessTransitions(AbstractStateMachine.this, context, state).subscribe(); - }; + public Mono doOnComplete(StateContext context) { + if (log.isDebugEnabled()) { + log.debug("State onComplete: state=[" + state + "] context=[" + context + "]"); + } + return ((AbstractStateMachine)getRelayStateMachine()).executeTriggerlessTransitions(AbstractStateMachine.this, context, state); + } }); if (state.isSubmachineState()) { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java index 78cc3aab..678c4c5f 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/ReactiveStateMachineExecutor.java @@ -56,7 +56,7 @@ import reactor.core.publisher.Mono; /** * Default reactive implementation of a {@link StateMachineExecutor}. - * + * * @author Janne Valkealahti * * @param the type of state @@ -168,6 +168,9 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i @Override public Mono executeTriggerlessTransitions(StateContext context, State state) { if (stateMachine.getState() != null) { + if (log.isDebugEnabled()) { + log.debug("About to handleTriggerlessTransitions"); + } return handleTriggerlessTransitions(context, state); } return Mono.empty(); diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/StateDoActivityActionTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/StateDoActivityActionTests.java index d1770524..05c8d302 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/StateDoActivityActionTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/action/StateDoActivityActionTests.java @@ -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)); diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/monitor/StateMachineMonitorTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/monitor/StateMachineMonitorTests.java index ed84795a..f9d31230 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/monitor/StateMachineMonitorTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/monitor/StateMachineMonitorTests.java @@ -195,7 +195,6 @@ public class StateMachineMonitorTests extends AbstractStateMachineTests { @Override public void action(StateMachine stateMachine, Function, Mono> action, long duration) { - System.out.println("XXX HI"); actions.put(action, new Actions(action, duration)); latch.countDown(); } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/state/CompletionEventTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/state/CompletionEventTests.java index dd9ab656..6e4d501f 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/state/CompletionEventTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/state/CompletionEventTests.java @@ -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")); } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/state/ObjectStateTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/state/ObjectStateTests.java new file mode 100644 index 00000000..9b8a4fc2 --- /dev/null +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/state/ObjectStateTests.java @@ -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 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 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 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 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 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 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 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 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 { + + AtomicInteger count = new AtomicInteger(); + + @Override + public Mono apply(StateContext context) { + return Mono.empty().doOnSuccess(d -> { + count.incrementAndGet(); + }); + } + } + + private static class TestBlockingAction implements ReactiveAction { + + AtomicInteger countBefore = new AtomicInteger(); + AtomicInteger countAfter = new AtomicInteger(); + AtomicInteger countInterrupt = new AtomicInteger(); + + @Override + public Mono apply(StateContext context) { + return Mono.fromRunnable(() -> { + countBefore.incrementAndGet(); + try { + Thread.sleep(1000); + } catch (Exception e) { + countInterrupt.incrementAndGet(); + } + countAfter.incrementAndGet(); + }); + } + } +} diff --git a/spring-statemachine-samples/ordershipping/src/test/java/demo/ordershipping/StateMachineTests.java b/spring-statemachine-samples/ordershipping/src/test/java/demo/ordershipping/StateMachineTests.java index 18121557..9de3d574 100644 --- a/spring-statemachine-samples/ordershipping/src/test/java/demo/ordershipping/StateMachineTests.java +++ b/spring-statemachine-samples/ordershipping/src/test/java/demo/ordershipping/StateMachineTests.java @@ -106,7 +106,7 @@ public class StateMachineTests { .setHeader("payment", "1000").build()) .expectStates("ORDER_SHIPPED") // .expectStateChanged(4) - .expectStateChanged(3) + .expectStateChanged(2) .expectStateMachineStopped(3) .and() .build();