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

Polish AbstractState

Browse Source 
- Polish previous change to AbstractState by using a shared function
  completionStateListenerSink to track completion sink as functionality
  between entries to submachine/regions should be similar.
- Mostly relates to #743
This commit is contained in:
Janne Valkealahti
2019-05-26 15:20:03 +01:00
parent 3cf07e8066
commit 10f986ce06
1 changed files with 22 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -252,26 +252,7 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
return Mono.defer(() -> {
if (submachine != null) {
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()) {
sink.success();
}
}
}
}
};
submachine.addStateListener(l);
});
})
.flatMap(submachine -> completionStateListenerSink(submachine))
// TODO: REACTOR this is causing cancel which breaks some things
// .then(handleStateDoOnComplete(context))
.then(Mono.fromRunnable(() -> notifyStateOnComplete(context)))
@@ -280,25 +261,7 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
} 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);
});
})
.flatMap(region -> completionStateListenerSink(region))
.then(handleStateDoOnComplete(context))
.then(Mono.fromRunnable(() -> notifyStateOnComplete(context)))
.subscribe();
@@ -471,6 +434,26 @@ public abstract class AbstractState<S, E> extends LifecycleObjectSupport impleme
}
}
private Mono<Void> completionStateListenerSink(Region<S, E> region) {
return Mono.create(sink -> {
final StateMachineListener<S, E> listener = 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(listener);
region.addStateListener(listener);
});
}
private void disposeDisposables() {
Disposable disposable;
while ((disposable = disposables.poll()) != null) {