From f1bf35e6883af292f0e60d3e8135d37303322f93 Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Sat, 3 Aug 2019 12:59:52 +0100 Subject: [PATCH] Trigger reactive changes - Triggers evaluate method is now reactive. - Various changes for signature change. - Fixes #799 --- .../support/AbstractStateMachine.java | 31 ++-- .../statemachine/support/CountTrigger.java | 153 ------------------ .../support/ReactiveStateMachineExecutor.java | 31 ++-- .../statemachine/trigger/EventTrigger.java | 8 +- .../statemachine/trigger/TimerTrigger.java | 33 ++-- .../statemachine/trigger/Trigger.java | 8 +- 6 files changed, 65 insertions(+), 199 deletions(-) delete mode 100644 spring-statemachine-core/src/main/java/org/springframework/statemachine/support/CountTrigger.java 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 b75e6e72..8b94f223 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 @@ -66,6 +66,7 @@ import org.springframework.statemachine.transition.TransitionConflictPolicy; import org.springframework.statemachine.transition.TransitionKind; import org.springframework.statemachine.trigger.DefaultTriggerContext; import org.springframework.statemachine.trigger.Trigger; +import org.springframework.statemachine.trigger.TriggerContext; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -643,6 +644,7 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo private Flux> acceptEvent(Message message) { return Flux.defer(() -> { + TriggerContext triggerContext = new DefaultTriggerContext(message.getPayload()); State cs = currentState; if (cs != null) { if (cs.shouldDefer(message)) { @@ -652,18 +654,23 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo return cs.sendEvent(message).collectList().flatMapMany(l -> { Flux> ret = Flux.fromIterable(l); if (!l.stream().anyMatch(er -> er.getResultType() == ResultType.ACCEPTED)) { - ret = ret.concatWith(Mono.defer(() -> { - for (Transition transition : transitions) { - State source = transition.getSource(); - Trigger trigger = transition.getTrigger(); - if (cs != null && StateMachineUtils.containsAtleastOne(source.getIds(), cs.getIds())) { - if (trigger != null && trigger.evaluate(new DefaultTriggerContext(message.getPayload()))) { - return stateMachineExecutor.queueEvent(Mono.just(message)).thenReturn(StateMachineEventResult.from(this, message, ResultType.ACCEPTED)); - } - } - } - return Mono.just(StateMachineEventResult.from(this, message, ResultType.DENIED)); - })); + Mono> result = Flux.fromIterable(transitions) + .filter(transition -> cs != null && transition.getTrigger() != null) + .filter(transition -> StateMachineUtils.containsAtleastOne(transition.getSource().getIds(), cs.getIds())) + .flatMap(transition -> { + return Mono.from(transition.getTrigger().evaluate(triggerContext)) + .flatMap(e -> { + if (e) { + return stateMachineExecutor.queueEvent(Mono.just(message)) + .thenReturn(StateMachineEventResult.from(this, message, ResultType.ACCEPTED)); + } else { + return Mono.empty(); + } + }); + }) + .next() + .switchIfEmpty(Mono.just(StateMachineEventResult.from(this, message, ResultType.DENIED))); + ret = ret.concatWith(result); } return ret; }); diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/CountTrigger.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/CountTrigger.java deleted file mode 100644 index 524b6a24..00000000 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/CountTrigger.java +++ /dev/null @@ -1,153 +0,0 @@ -/* - * Copyright 2016 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.support; - -import java.util.Date; -import java.util.concurrent.TimeUnit; - -import org.springframework.scheduling.Trigger; -import org.springframework.scheduling.TriggerContext; -import org.springframework.scheduling.support.PeriodicTrigger; -import org.springframework.util.Assert; - -/** - * Enhanced implementation following same logic from {@link PeriodicTrigger} - * except also adding a counter how many times a trigger can fire. If given - * count is either zero on a negative value, counter functionality is disabled. - * - * @author Janne Valkealahti - * @see PeriodicTrigger - */ -public class CountTrigger implements Trigger { - - private final int count; - private final long period; - private final TimeUnit timeUnit; - private volatile long initialDelay = 0; - private volatile boolean fixedRate = false; - private volatile int counter = 0; - - /** - * Create a trigger with the given period in milliseconds and firing - * exactly one time. - * - * @param period the period - */ - public CountTrigger(long period) { - this(1, period, null); - } - - /** - * Create a trigger with the given count, period and time unit. The time unit will - * apply not only to the period but also to any 'initialDelay' value, if - * configured on this Trigger later via {@link #setInitialDelay(long)}. - * - * @param count the count - * @param period the period - * @param timeUnit the time unit - */ - public CountTrigger(int count, long period, TimeUnit timeUnit) { - this(count, period, 0, timeUnit); - } - - /** - * Create a trigger with the given count, period and time unit. The time unit will - * apply not only to the period but also to any 'initialDelay' value, if - * configured on this Trigger later via {@link #setInitialDelay(long)}. - * - * @param count the count - * @param period the period - * @param timeUnit the time unit - * @param initialDelay the initial delay - */ - public CountTrigger(int count, long period, long initialDelay, TimeUnit timeUnit) { - Assert.isTrue(period >= 0, "period must not be negative"); - Assert.isTrue(count >= 0, "count must not be negative"); - this.timeUnit = (timeUnit != null ? timeUnit : TimeUnit.MILLISECONDS); - this.period = this.timeUnit.toMillis(period); - this.count = count; - setInitialDelay(initialDelay); - } - - /** - * Specify the delay for the initial execution. It will be evaluated in - * terms of this trigger's {@link TimeUnit}. If no time unit was explicitly - * provided upon instantiation, the default is milliseconds. - * - * @param initialDelay the new initial delay - */ - public void setInitialDelay(long initialDelay) { - this.initialDelay = this.timeUnit.toMillis(initialDelay); - } - - /** - * Specify whether the periodic interval should be measured between the - * scheduled start times rather than between actual completion times. - * The latter, "fixed delay" behavior, is the default. - * - * @param fixedRate the new fixed rate - */ - public void setFixedRate(boolean fixedRate) { - this.fixedRate = fixedRate; - } - - @Override - public Date nextExecutionTime(TriggerContext triggerContext) { - if (count > 0) { - if (++counter > count) { - return null; - } - } - if (triggerContext.lastScheduledExecutionTime() == null) { - return new Date(System.currentTimeMillis() + this.initialDelay); - } - else if (this.fixedRate) { - return new Date(triggerContext.lastScheduledExecutionTime().getTime() + this.period); - } - return new Date(triggerContext.lastCompletionTime().getTime() + this.period); - } - - @Override - public int hashCode() { - final int prime = 31; - int result = 1; - result = prime * result + count; - result = prime * result + (fixedRate ? 1231 : 1237); - result = prime * result + (int) (initialDelay ^ (initialDelay >>> 32)); - result = prime * result + (int) (period ^ (period >>> 32)); - return result; - } - - @Override - public boolean equals(Object obj) { - if (this == obj) - return true; - if (obj == null) - return false; - if (getClass() != obj.getClass()) - return false; - CountTrigger other = (CountTrigger) obj; - if (count != other.count) - return false; - if (fixedRate != other.fixedRate) - return false; - if (initialDelay != other.initialDelay) - return false; - if (period != other.period) - return false; - return true; - } -} 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 575e62fb..137af75d 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 @@ -44,6 +44,7 @@ import org.springframework.statemachine.transition.AbstractTransition; import org.springframework.statemachine.transition.Transition; import org.springframework.statemachine.transition.TransitionConflictPolicy; import org.springframework.statemachine.trigger.DefaultTriggerContext; +import org.springframework.statemachine.trigger.TriggerContext; import org.springframework.statemachine.trigger.TimerTrigger; import org.springframework.statemachine.trigger.Trigger; import org.springframework.statemachine.trigger.TriggerListener; @@ -214,18 +215,24 @@ public class ReactiveStateMachineExecutor extends LifecycleObjectSupport i log.info("Current state " + currentState + " deferred event " + queuedEvent); return Mono.just(new TriggerQueueItem(null, queuedEvent)); } - for (Transition transition : transitions) { - State source = transition.getSource(); - Trigger trigger = transition.getTrigger(); - - if (StateMachineUtils.containsAtleastOne(source.getIds(), currentState.getIds())) { - if (trigger != null && trigger.evaluate(new DefaultTriggerContext(queuedEvent.getPayload()))) { - deferList.remove(queuedEvent); - return Mono.just(new TriggerQueueItem(trigger, queuedEvent)); - } - } - } - return Mono.empty(); + TriggerContext triggerContext = new DefaultTriggerContext(queuedEvent.getPayload()); + return Flux.fromIterable(transitions) + .filter(transition -> transition.getTrigger() != null) + .filter(transition -> StateMachineUtils.containsAtleastOne(transition.getSource().getIds(), + currentState.getIds())) + .flatMap(transition -> { + return Mono.from(transition.getTrigger().evaluate(triggerContext)) + .flatMap(e -> { + if (e) { + return Mono.just(transition.getTrigger()); + } else { + return Mono.empty(); + } + }); + }) + .next() + .doOnNext(trigger -> deferList.remove(queuedEvent)) + .map(trigger -> new TriggerQueueItem(trigger, queuedEvent)); }); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/EventTrigger.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/EventTrigger.java index 6c46c6ae..9d1a0de8 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/EventTrigger.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/EventTrigger.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,8 @@ package org.springframework.statemachine.trigger; import org.springframework.util.ObjectUtils; +import reactor.core.publisher.Mono; + public class EventTrigger implements Trigger { private final E event; @@ -26,8 +28,8 @@ public class EventTrigger implements Trigger { } @Override - public boolean evaluate(TriggerContext context) { - return ObjectUtils.nullSafeEquals(event, context.getEvent()); + public Mono evaluate(TriggerContext context) { + return Mono.just(ObjectUtils.nullSafeEquals(event, context.getEvent())); } @Override diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/TimerTrigger.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/TimerTrigger.java index 92d497ec..a36c4b9e 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/TimerTrigger.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/TimerTrigger.java @@ -15,12 +15,12 @@ */ package org.springframework.statemachine.trigger; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; +import java.time.Duration; -import org.springframework.statemachine.support.CountTrigger; import org.springframework.statemachine.support.LifecycleObjectSupport; +import reactor.core.Disposable; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; /** @@ -37,7 +37,7 @@ public class TimerTrigger extends LifecycleObjectSupport implements Trigge private final CompositeTriggerListener triggerListener = new CompositeTriggerListener(); private final long period; private final int count; - private volatile ScheduledFuture scheduled; + private Disposable disposable; /** * Instantiates a new timer trigger. @@ -68,8 +68,8 @@ public class TimerTrigger extends LifecycleObjectSupport implements Trigge } @Override - public boolean evaluate(TriggerContext context) { - return false; + public Mono evaluate(TriggerContext context) { + return Mono.just(false); } @Override @@ -102,7 +102,7 @@ public class TimerTrigger extends LifecycleObjectSupport implements Trigge @Override public void arm() { - if (scheduled != null) { + if (disposable != null) { return; } schedule(); @@ -117,13 +117,14 @@ public class TimerTrigger extends LifecycleObjectSupport implements Trigge private void schedule() { long initialDelay = count > 0 ? period : 0; - scheduled = getTaskScheduler().schedule(new Runnable() { - - @Override - public void run() { + Flux interval = Flux.interval(Duration.ofMillis(initialDelay), Duration.ofMillis(period)) + .doOnNext(c -> { notifyTriggered(); - } - }, new CountTrigger(count, period, initialDelay, TimeUnit.MILLISECONDS)); + }); + if (count > 0) { + interval = interval.take(count); + } + disposable = interval.subscribe(); } private void notifyTriggered() { @@ -131,9 +132,9 @@ public class TimerTrigger extends LifecycleObjectSupport implements Trigge } private void cancel() { - if (scheduled != null) { - scheduled.cancel(true); + if (disposable != null) { + disposable.dispose(); } - scheduled = null; + disposable = null; } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/Trigger.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/Trigger.java index ec3871b4..a62baab7 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/Trigger.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/trigger/Trigger.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 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.trigger; import org.springframework.statemachine.transition.Transition; +import reactor.core.publisher.Mono; + /** * {@code Trigger} is the cause of the {@link Transition}. Cause is usually an * event but can be some other signal or a change in some condition. @@ -32,9 +34,9 @@ public interface Trigger { * Evaluate trigger. * * @param context the context - * @return true, triggers is fired, false otherwise + * @return Mono for completion with true, if trigger is fired, false otherwise */ - boolean evaluate(TriggerContext context); + Mono evaluate(TriggerContext context); /** * Adds the trigger listener.