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();