Skip to content

Commit b064913

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

2 files changed

Lines changed: 94 additions & 100 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: 85 additions & 100 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,7 +132,19 @@ 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)
@@ -157,16 +156,23 @@ private void cleanUp(WorkflowModel result, Throwable ex) {
157156
private CompletableFuture<WorkflowModel> handleException(Throwable exception) {
158157
final Throwable cause =
159158
exception instanceof CompletionException ? exception.getCause() : exception;
160-
if (!(cause instanceof CancellationException)) {
159+
if (cause instanceof CancellationException) {
160+
return CompletableFuture.failedFuture(cause);
161+
} else {
161162
return status(WorkflowStatus.FAULTED)
162163
.thenCompose(
163-
__ ->
164+
changed ->
164165
publishEvent(
166+
changed,
165167
workflowContext,
166168
l -> l.onWorkflowFailed(new WorkflowFailedEvent(workflowContext, cause))))
167-
.thenCompose(__ -> CompletableFuture.failedFuture(exception));
169+
.thenCompose(
170+
failed ->
171+
CompletableFuture.failedFuture(
172+
failed
173+
? exception
174+
: new CancellationException("Workflow has been cancelled")));
168175
}
169-
return CompletableFuture.failedFuture(exception);
170176
}
171177

172178
private WorkflowModel filterAndValidate(WorkflowModel model) {
@@ -232,45 +238,43 @@ public <T> T outputAs(Class<T> clazz) {
232238
: null;
233239
}
234240

235-
public CompletableFuture<Boolean> status(WorkflowStatus state) {
241+
public CompletableFuture<Boolean> status(WorkflowStatus requestedState) {
236242
WorkflowStatus prevState;
243+
WorkflowStatus state;
244+
statusLock.lock();
237245
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);
246+
state = prevState = this.status.get();
247+
if (prevState != WorkflowStatus.CANCELLED
248+
&& prevState != WorkflowStatus.COMPLETED
249+
&& prevState != WorkflowStatus.FAULTED) {
250+
state = requestedState;
251+
status.set(requestedState);
248252
}
249-
this.status.set(state);
250253
} finally {
251254
statusLock.unlock();
252255
}
253256
return publishStatusChange(prevState, state);
254257
}
255258

256-
private static boolean isTerminal(WorkflowStatus state) {
257-
return state == WorkflowStatus.COMPLETED || state == WorkflowStatus.FAULTED;
258-
}
259-
260259
protected final void setStatus(WorkflowStatus state) {
261260
this.status.set(state);
262261
}
263262

263+
private CompletableFuture<Boolean> publishStatusChangeRequest(
264+
WorkflowStatus prevState, WorkflowStatus state) {
265+
CompletableFuture<Boolean> future = publishStatusChange(prevState, state);
266+
if (!future.isDone()) {
267+
outOfOrderListeners.add(future);
268+
}
269+
return future;
270+
}
271+
264272
private CompletableFuture<Boolean> publishStatusChange(
265273
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);
274+
return publishEvent(
275+
prevState != state,
276+
workflowContext,
277+
l -> l.onWorkflowStatusChanged(new WorkflowStatusEvent(workflowContext, prevState, state)));
274278
}
275279

276280
@Override
@@ -291,7 +295,7 @@ public boolean suspend() {
291295
WorkflowStatus prevState = internalSuspend();
292296
boolean result = prevState != WorkflowStatus.SUSPENDED;
293297
if (result) {
294-
publishStatusChange(prevState, WorkflowStatus.SUSPENDED)
298+
publishStatusChangeRequest(prevState, WorkflowStatus.SUSPENDED)
295299
.thenCompose(
296300
__ ->
297301
publishEvent(
@@ -305,7 +309,7 @@ public boolean suspend() {
305309
public CompletableFuture<Boolean> suspendFuture() {
306310
WorkflowStatus prevState = internalSuspend();
307311
return prevState != WorkflowStatus.SUSPENDED
308-
? publishStatusChange(prevState, WorkflowStatus.SUSPENDED)
312+
? publishStatusChangeRequest(prevState, WorkflowStatus.SUSPENDED)
309313
.thenCompose(
310314
__ ->
311315
publishEvent(
@@ -338,7 +342,7 @@ public boolean resume() {
338342
WorkflowStatus prevStatus = internalResume();
339343
boolean result = prevStatus != WorkflowStatus.RUNNING;
340344
if (result) {
341-
publishStatusChange(prevStatus, WorkflowStatus.RUNNING)
345+
publishStatusChangeRequest(prevStatus, WorkflowStatus.RUNNING)
342346
.thenCompose(
343347
__ ->
344348
publishEvent(
@@ -352,7 +356,7 @@ public boolean resume() {
352356
public CompletableFuture<Boolean> resumeFuture() {
353357
WorkflowStatus prevStatus = internalResume();
354358
return prevStatus != WorkflowStatus.RUNNING
355-
? publishStatusChange(prevStatus, WorkflowStatus.RUNNING)
359+
? publishStatusChangeRequest(prevStatus, WorkflowStatus.RUNNING)
356360
.thenCompose(
357361
__ ->
358362
publishEvent(
@@ -382,13 +386,12 @@ private WorkflowStatus internalResume() {
382386
return result;
383387
}
384388

385-
public CompletableFuture<TaskContext> cancelCheck(TaskContext t) {
389+
public <T> CompletableFuture<T> cancelCheck(T t) {
386390
try {
387391
statusLock.lock();
388392
if (status.get() == WorkflowStatus.CANCELLED) {
389-
CompletableFuture<TaskContext> cancelled = new CompletableFuture<TaskContext>();
390-
cancelled.completeExceptionally(
391-
new CancellationException("Task " + t.taskName() + " has been cancelled"));
393+
CompletableFuture<T> cancelled = new CompletableFuture<>();
394+
cancelled.completeExceptionally(new CancellationException(t + " has been cancelled"));
392395
return cancelled;
393396
}
394397
} finally {
@@ -419,68 +422,50 @@ public CompletableFuture<TaskContext> suspendedCheck(TaskContext t) {
419422

420423
@Override
421424
public boolean cancel() {
422-
CancelRequest request = internalCancel();
423-
if (request.accepted()) {
424-
publishCancelled(request);
425+
WorkflowStatus prevStatus = internalCancel();
426+
boolean result = prevStatus != WorkflowStatus.CANCELLED;
427+
if (result) {
428+
publishStatusChangeRequest(prevStatus, WorkflowStatus.CANCELLED)
429+
.thenCompose(
430+
changed ->
431+
publishEvent(
432+
workflowContext,
433+
l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext))));
425434
}
426-
return request.accepted();
435+
return result;
427436
}
428437

429438
@Override
430439
public CompletableFuture<Boolean> cancelFuture() {
431-
CancelRequest request = internalCancel();
432-
return request.accepted()
433-
? publishCancelled(request).thenApply(__ -> true)
434-
: CompletableFuture.completedFuture(false);
435-
}
436-
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)
440+
WorkflowStatus prevState = internalCancel();
441+
return publishStatusChangeRequest(prevState, WorkflowStatus.CANCELLED)
445442
.thenCompose(
446-
__ ->
443+
changed ->
447444
publishEvent(
445+
changed,
448446
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-
}
447+
l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext))));
464448
}
465449

466-
private CancelRequest internalCancel() {
450+
private WorkflowStatus internalCancel() {
451+
WorkflowStatus result;
452+
Collection<CompletableFuture<?>> toCancel = null;
467453
try {
468454
statusLock.lock();
469455
if (TaskExecutorHelper.isActive(status.get())) {
470-
Collection<CompletableFuture<?>> toCancel = new ArrayList<>(cancelables);
456+
toCancel = new ArrayList<>(cancelables);
471457
cancelables.clear();
472-
CompletableFuture<Void> published = new CompletableFuture<>();
473-
cancelPublished = published;
474-
return new CancelRequest(status.getAndSet(WorkflowStatus.CANCELLED), toCancel, published);
458+
result = status.getAndSet(WorkflowStatus.CANCELLED);
475459
} else {
476-
return new CancelRequest(
477-
WorkflowStatus.CANCELLED,
478-
Collections.emptyList(),
479-
CompletableFuture.completedFuture(null));
460+
result = WorkflowStatus.CANCELLED;
480461
}
481462
} finally {
482463
statusLock.unlock();
483464
}
465+
if (result != WorkflowStatus.CANCELLED && toCancel != null) {
466+
toCancel.forEach(t -> t.cancel(true));
467+
}
468+
return result;
484469
}
485470

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

0 commit comments

Comments
 (0)