Skip to content

Commit 5fbf2a0

Browse files
fjtiradofjtirado
committed
[Fix #1749] Alternative approach
Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>
1 parent 8085c7b commit 5fbf2a0

2 files changed

Lines changed: 98 additions & 112 deletions

File tree

‎impl/core/src/main/java/io/serverlessworkflow/impl/LifecycleEventsUtils.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,15 @@ public class LifecycleEventsUtils {
2828

2929
private LifecycleEventsUtils() {}
3030

31+
public static CompletableFuture<Boolean> publishEvent(
32+
boolean statusChanged,
33+
WorkflowContext workflowContext,
34+
Function<WorkflowExecutionCompletableListener, CompletableFuture<?>> function) {
35+
return statusChanged
36+
? publishEvent(workflowContext, function).thenApply(__ -> true)
37+
: CompletableFuture.completedFuture(false);
38+
}
39+
3140
public static CompletableFuture<?> publishEvent(
3241
WorkflowContext workflowContext,
3342
Function<WorkflowExecutionCompletableListener, CompletableFuture<?>> function) {

‎impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java‎

Lines changed: 89 additions & 112 deletions
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ public class WorkflowMutableInstance implements WorkflowInstance {
6363

6464
private Collection<CompletableFuture<?>> cancelables = new ArrayList<>();
6565

66-
private volatile CompletableFuture<?> cancelPublished = CompletableFuture.completedFuture(null);
66+
private Collection<CompletableFuture<Boolean>> outOfOrderListeners = new ArrayList<>();
6767

6868
protected WorkflowMutableInstance(WorkflowDefinition definition, String id, WorkflowModel input) {
6969
this.id = id;
@@ -107,35 +107,22 @@ protected final CompletableFuture<WorkflowModel> startExecution(
107107
.orElse(input))
108108
.whenComplete(this::setCompleteDate)
109109
.thenApply(this::filterAndValidate)
110-
.thenCompose(this::publishEvents)
110+
.thenCompose(this::cancelCheck)
111+
.thenCompose(this::publishCompletionEvents)
111112
.exceptionallyCompose(this::handleException))
112-
.handle(this::cleanUpAfterCancelPublished)
113+
.handle(this::cleanUpWaitingListeners)
113114
.thenCompose(Function.identity());
114115
futureRef.set(future);
115116
}
116117
return future;
117118
}
118119

119-
private CompletableFuture<WorkflowModel> cleanUpAfterCancelPublished(
120-
WorkflowModel result, Throwable ex) {
121-
return cancelPublished
122-
.handle(
123-
(__, ___) -> {
124-
cleanUp(result, ex);
125-
return null;
126-
})
127-
.thenCompose(
128-
__ ->
129-
ex == null
130-
? CompletableFuture.completedFuture(result)
131-
: CompletableFuture.failedFuture(ex));
132-
}
133-
134-
private CompletableFuture<WorkflowModel> publishEvents(WorkflowModel model) {
120+
private CompletableFuture<WorkflowModel> publishCompletionEvents(WorkflowModel model) {
135121
return status(WorkflowStatus.COMPLETED)
136122
.thenCompose(
137-
__ ->
123+
changed ->
138124
publishEvent(
125+
changed,
139126
workflowContext,
140127
l -> l.onWorkflowCompleted(new WorkflowCompletedEvent(workflowContext, model))))
141128
.thenApply(__ -> model);
@@ -145,28 +132,48 @@ private void setCompleteDate(WorkflowModel result, Throwable ex) {
145132
completedAt = Instant.now();
146133
}
147134

148-
private void cleanUp(WorkflowModel result, Throwable ex) {
135+
private CompletableFuture<WorkflowModel> cleanUpWaitingListeners(
136+
WorkflowModel model, Throwable ex) {
137+
return CompletableFuture.allOf(
138+
outOfOrderListeners.toArray(new CompletableFuture[outOfOrderListeners.size()]))
139+
.whenComplete((__, ___) -> cleanUp())
140+
.thenCompose(
141+
__ ->
142+
ex == null
143+
? CompletableFuture.completedFuture(model)
144+
: CompletableFuture.failedFuture(ex));
145+
}
146+
147+
private void cleanUp() {
149148
additionalObjects.values().stream()
150149
.filter(AutoCloseable.class::isInstance)
151150
.map(AutoCloseable.class::cast)
152151
.forEach(WorkflowUtils::safeClose);
153152
additionalObjects.clear();
154153
workflowContext.definition().removeInstance(this);
154+
outOfOrderListeners.clear();
155155
}
156156

157157
private CompletableFuture<WorkflowModel> handleException(Throwable exception) {
158158
final Throwable cause =
159159
exception instanceof CompletionException ? exception.getCause() : exception;
160-
if (!(cause instanceof CancellationException)) {
160+
if (cause instanceof CancellationException) {
161+
return CompletableFuture.failedFuture(cause);
162+
} else {
161163
return status(WorkflowStatus.FAULTED)
162164
.thenCompose(
163-
__ ->
165+
changed ->
164166
publishEvent(
167+
changed,
165168
workflowContext,
166169
l -> l.onWorkflowFailed(new WorkflowFailedEvent(workflowContext, cause))))
167-
.thenCompose(__ -> CompletableFuture.failedFuture(exception));
170+
.thenCompose(
171+
failed ->
172+
CompletableFuture.failedFuture(
173+
failed
174+
? exception
175+
: new CancellationException("Workflow has been cancelled")));
168176
}
169-
return CompletableFuture.failedFuture(exception);
170177
}
171178

172179
private WorkflowModel filterAndValidate(WorkflowModel model) {
@@ -232,45 +239,43 @@ public <T> T outputAs(Class<T> clazz) {
232239
: null;
233240
}
234241

235-
public CompletableFuture<Boolean> status(WorkflowStatus state) {
242+
public CompletableFuture<Boolean> status(WorkflowStatus requestedState) {
236243
WorkflowStatus prevState;
244+
WorkflowStatus state;
245+
statusLock.lock();
237246
try {
238-
statusLock.lock();
239-
prevState = this.status.get();
240-
if (prevState == WorkflowStatus.CANCELLED) {
241-
// cancellation is final: late callbacks (e.g. an event received by a listen task) must
242-
// not move the instance out of it, and a terminal transition must fail so that neither
243-
// onWorkflowCompleted nor onWorkflowFailed is published for a cancelled instance
244-
return isTerminal(state)
245-
? CompletableFuture.failedFuture(
246-
new CancellationException("Workflow instance " + id + " has been cancelled"))
247-
: CompletableFuture.completedFuture(false);
247+
state = prevState = this.status.get();
248+
if (prevState != WorkflowStatus.CANCELLED
249+
&& prevState != WorkflowStatus.COMPLETED
250+
&& prevState != WorkflowStatus.FAULTED) {
251+
state = requestedState;
252+
status.set(requestedState);
248253
}
249-
this.status.set(state);
250254
} finally {
251255
statusLock.unlock();
252256
}
253257
return publishStatusChange(prevState, state);
254258
}
255259

256-
private static boolean isTerminal(WorkflowStatus state) {
257-
return state == WorkflowStatus.COMPLETED || state == WorkflowStatus.FAULTED;
258-
}
259-
260260
protected final void setStatus(WorkflowStatus state) {
261261
this.status.set(state);
262262
}
263263

264+
private CompletableFuture<Boolean> publishStatusChangeRequest(
265+
WorkflowStatus prevState, WorkflowStatus state) {
266+
CompletableFuture<Boolean> future = publishStatusChange(prevState, state);
267+
if (!future.isDone()) {
268+
outOfOrderListeners.add(future);
269+
}
270+
return future;
271+
}
272+
264273
private CompletableFuture<Boolean> publishStatusChange(
265274
WorkflowStatus prevState, WorkflowStatus state) {
266-
return prevState != state
267-
? publishEvent(
268-
workflowContext,
269-
l ->
270-
l.onWorkflowStatusChanged(
271-
new WorkflowStatusEvent(workflowContext, prevState, state)))
272-
.thenApply(__ -> true)
273-
: CompletableFuture.completedFuture(false);
275+
return publishEvent(
276+
prevState != state,
277+
workflowContext,
278+
l -> l.onWorkflowStatusChanged(new WorkflowStatusEvent(workflowContext, prevState, state)));
274279
}
275280

276281
@Override
@@ -291,7 +296,7 @@ public boolean suspend() {
291296
WorkflowStatus prevState = internalSuspend();
292297
boolean result = prevState != WorkflowStatus.SUSPENDED;
293298
if (result) {
294-
publishStatusChange(prevState, WorkflowStatus.SUSPENDED)
299+
publishStatusChangeRequest(prevState, WorkflowStatus.SUSPENDED)
295300
.thenCompose(
296301
__ ->
297302
publishEvent(
@@ -305,7 +310,7 @@ public boolean suspend() {
305310
public CompletableFuture<Boolean> suspendFuture() {
306311
WorkflowStatus prevState = internalSuspend();
307312
return prevState != WorkflowStatus.SUSPENDED
308-
? publishStatusChange(prevState, WorkflowStatus.SUSPENDED)
313+
? publishStatusChangeRequest(prevState, WorkflowStatus.SUSPENDED)
309314
.thenCompose(
310315
__ ->
311316
publishEvent(
@@ -338,7 +343,7 @@ public boolean resume() {
338343
WorkflowStatus prevStatus = internalResume();
339344
boolean result = prevStatus != WorkflowStatus.RUNNING;
340345
if (result) {
341-
publishStatusChange(prevStatus, WorkflowStatus.RUNNING)
346+
publishStatusChangeRequest(prevStatus, WorkflowStatus.RUNNING)
342347
.thenCompose(
343348
__ ->
344349
publishEvent(
@@ -351,15 +356,13 @@ public boolean resume() {
351356
@Override
352357
public CompletableFuture<Boolean> resumeFuture() {
353358
WorkflowStatus prevStatus = internalResume();
354-
return prevStatus != WorkflowStatus.RUNNING
355-
? publishStatusChange(prevStatus, WorkflowStatus.RUNNING)
356-
.thenCompose(
357-
__ ->
358-
publishEvent(
359-
workflowContext,
360-
l -> l.onWorkflowResumed(new WorkflowResumedEvent(workflowContext))))
361-
.thenApply(__ -> true)
362-
: CompletableFuture.completedFuture(false);
359+
return publishStatusChangeRequest(prevStatus, WorkflowStatus.RUNNING)
360+
.thenCompose(
361+
change ->
362+
publishEvent(
363+
change,
364+
workflowContext,
365+
l -> l.onWorkflowResumed(new WorkflowResumedEvent(workflowContext))));
363366
}
364367

365368
private WorkflowStatus internalResume() {
@@ -382,13 +385,12 @@ private WorkflowStatus internalResume() {
382385
return result;
383386
}
384387

385-
public CompletableFuture<TaskContext> cancelCheck(TaskContext t) {
388+
public <T> CompletableFuture<T> cancelCheck(T t) {
386389
try {
387390
statusLock.lock();
388391
if (status.get() == WorkflowStatus.CANCELLED) {
389-
CompletableFuture<TaskContext> cancelled = new CompletableFuture<TaskContext>();
390-
cancelled.completeExceptionally(
391-
new CancellationException("Task " + t.taskName() + " has been cancelled"));
392+
CompletableFuture<T> cancelled = new CompletableFuture<>();
393+
cancelled.completeExceptionally(new CancellationException(t + " has been cancelled"));
392394
return cancelled;
393395
}
394396
} finally {
@@ -419,68 +421,43 @@ public CompletableFuture<TaskContext> suspendedCheck(TaskContext t) {
419421

420422
@Override
421423
public boolean cancel() {
422-
CancelRequest request = internalCancel();
423-
if (request.accepted()) {
424-
publishCancelled(request);
425-
}
426-
return request.accepted();
424+
CompletableFuture<Boolean> result = internalCancel();
425+
return !result.isDone() || result.join();
427426
}
428427

429428
@Override
430429
public CompletableFuture<Boolean> cancelFuture() {
431-
CancelRequest request = internalCancel();
432-
return request.accepted()
433-
? publishCancelled(request).thenApply(__ -> true)
434-
: CompletableFuture.completedFuture(false);
430+
return internalCancel();
435431
}
436432

437-
/**
438-
* Publishes the cancellation and only then cancels the pending futures. Cancelling them may
439-
* complete the execution pipeline synchronously, and the pipeline may also end on its own while
440-
* the cancellation is being published, so {@link #cleanUpAfterCancelPublished} waits for {@link
441-
* CancelRequest#published()} before clearing the instance metadata.
442-
*/
443-
private CompletableFuture<?> publishCancelled(CancelRequest request) {
444-
return publishStatusChange(request.prevStatus(), WorkflowStatus.CANCELLED)
445-
.thenCompose(
446-
__ ->
447-
publishEvent(
448-
workflowContext,
449-
l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext))))
450-
.whenComplete(
451-
(__, ex) -> {
452-
request.published().complete(null);
453-
request.toCancel().forEach(t -> t.cancel(true));
454-
});
455-
}
456-
457-
private record CancelRequest(
458-
WorkflowStatus prevStatus,
459-
Collection<CompletableFuture<?>> toCancel,
460-
CompletableFuture<Void> published) {
461-
boolean accepted() {
462-
return prevStatus != WorkflowStatus.CANCELLED;
463-
}
464-
}
465-
466-
private CancelRequest internalCancel() {
433+
private CompletableFuture<Boolean> internalCancel() {
434+
WorkflowStatus prevState;
435+
Collection<CompletableFuture<?>> toCancel = null;
467436
try {
468437
statusLock.lock();
469438
if (TaskExecutorHelper.isActive(status.get())) {
470-
Collection<CompletableFuture<?>> toCancel = new ArrayList<>(cancelables);
439+
toCancel = new ArrayList<>(cancelables);
471440
cancelables.clear();
472-
CompletableFuture<Void> published = new CompletableFuture<>();
473-
cancelPublished = published;
474-
return new CancelRequest(status.getAndSet(WorkflowStatus.CANCELLED), toCancel, published);
441+
prevState = status.getAndSet(WorkflowStatus.CANCELLED);
475442
} else {
476-
return new CancelRequest(
477-
WorkflowStatus.CANCELLED,
478-
Collections.emptyList(),
479-
CompletableFuture.completedFuture(null));
443+
prevState = WorkflowStatus.CANCELLED;
480444
}
481445
} finally {
482446
statusLock.unlock();
483447
}
448+
CompletableFuture<Boolean> result =
449+
publishStatusChangeRequest(prevState, WorkflowStatus.CANCELLED)
450+
.thenCompose(
451+
changed ->
452+
publishEvent(
453+
changed,
454+
workflowContext,
455+
l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext))));
456+
457+
if (prevState != WorkflowStatus.CANCELLED && toCancel != null) {
458+
toCancel.forEach(t -> t.cancel(true));
459+
}
460+
return result;
484461
}
485462

486463
public void addCancelable(CompletableFuture<?> cancelable) {

0 commit comments

Comments
 (0)