-
Notifications
You must be signed in to change notification settings - Fork 248
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor: introduce RetryProcessor (#4787)
* refactor: introduce retry processor * pr remarks
- Loading branch information
Showing
20 changed files
with
1,188 additions
and
131 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
61 changes: 61 additions & 0 deletions
61
...-lib/src/main/java/org/eclipse/edc/statemachine/retry/processor/EntityStateException.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| /* | ||
| * Copyright (c) 2025 Cofinity-X | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Apache License, Version 2.0 which is available at | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * Contributors: | ||
| * Cofinity-X - initial API and implementation | ||
| * | ||
| */ | ||
|
|
||
| package org.eclipse.edc.statemachine.retry.processor; | ||
|
|
||
| import org.eclipse.edc.spi.entity.StatefulEntity; | ||
| import org.jetbrains.annotations.NotNull; | ||
|
|
||
| /** | ||
| * Exception that describes generic failure in a process | ||
| */ | ||
| public class EntityStateException extends RuntimeException { | ||
|
|
||
| private final StatefulEntity<?> entity; | ||
| private final String processName; | ||
|
|
||
| public EntityStateException(StatefulEntity<?> entity, String processName, String message) { | ||
| super(message); | ||
| this.entity = entity; | ||
| this.processName = processName; | ||
| } | ||
|
|
||
| public StatefulEntity<?> getEntity() { | ||
| return entity; | ||
| } | ||
|
|
||
| public String getProcessName() { | ||
| return processName; | ||
| } | ||
|
|
||
| @NotNull String getRetryLimitExceededMessage() { | ||
| return "%s: ID %s. Attempt #%d failed to %s. Retry limit exceeded. Cause: %s".formatted( | ||
| entity.getClass().getSimpleName(), | ||
| entity.getId(), | ||
| entity.getStateCount(), | ||
| getProcessName(), | ||
| getMessage() | ||
| ); | ||
| } | ||
|
|
||
| @NotNull String getRetryFailedMessage() { | ||
| return "%s: ID %s. Attempt #%d failed to %s. Cause: %s".formatted( | ||
| entity.getClass().getSimpleName(), | ||
| entity.getId(), | ||
| entity.getStateCount(), | ||
| getProcessName(), | ||
| getMessage() | ||
| ); | ||
| } | ||
| } |
65 changes: 65 additions & 0 deletions
65
.../src/main/java/org/eclipse/edc/statemachine/retry/processor/FutureResultRetryProcess.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| /* | ||
| * Copyright (c) 2025 Cofinity-X | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Apache License, Version 2.0 which is available at | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * Contributors: | ||
| * Cofinity-X - initial API and implementation | ||
| * | ||
| */ | ||
|
|
||
| package org.eclipse.edc.statemachine.retry.processor; | ||
|
|
||
| import org.eclipse.edc.spi.entity.StatefulEntity; | ||
| import org.eclipse.edc.spi.response.StatusResult; | ||
|
|
||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.function.BiFunction; | ||
| import java.util.function.Function; | ||
|
|
||
| import static java.util.concurrent.CompletableFuture.failedFuture; | ||
|
|
||
| /** | ||
| * Process implementation that handles a process that returns {@link CompletableFuture} with enclosed {@link StatusResult} | ||
| * | ||
| * @param <E> entity type. | ||
| * @param <I> process input type. | ||
| * @param <O> process output type. | ||
| */ | ||
| public class FutureResultRetryProcess<E extends StatefulEntity<E>, I, O> implements Process<E, I, O> { | ||
|
|
||
| private final String name; | ||
| private final BiFunction<E, I, CompletableFuture<StatusResult<O>>> function; | ||
| private Function<String, E> entityReload; | ||
|
|
||
| public FutureResultRetryProcess(String name, BiFunction<E, I, CompletableFuture<StatusResult<O>>> function) { | ||
| this.name = name; | ||
| this.function = function; | ||
| } | ||
|
|
||
| @Override | ||
| public CompletableFuture<ProcessContext<E, O>> execute(ProcessContext<E, I> context) { | ||
| try { | ||
| return new FutureRetryProcess<>(name, function).entityReload(entityReload) | ||
| .execute(context) | ||
| .thenCompose(asyncContext -> new ResultRetryProcess<E, I, O>(name, (e, c) -> asyncContext.content()) | ||
| .execute(new ProcessContext<>(asyncContext.entity(), context.content()))); | ||
| } catch (Throwable throwable) { | ||
| return failedFuture(new UnrecoverableEntityStateException(reloadEntity(context.entity()), name, throwable.getMessage())); | ||
| } | ||
| } | ||
|
|
||
| public FutureResultRetryProcess<E, I, O> entityReload(Function<String, E> entityReload) { | ||
| this.entityReload = entityReload; | ||
| return this; | ||
| } | ||
|
|
||
| private E reloadEntity(E entity) { | ||
| return entityReload == null ? entity : entityReload.apply(entity.getId()); | ||
| } | ||
|
|
||
| } |
69 changes: 69 additions & 0 deletions
69
...ne-lib/src/main/java/org/eclipse/edc/statemachine/retry/processor/FutureRetryProcess.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| /* | ||
| * Copyright (c) 2025 Cofinity-X | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Apache License, Version 2.0 which is available at | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * Contributors: | ||
| * Cofinity-X - initial API and implementation | ||
| * | ||
| */ | ||
|
|
||
| package org.eclipse.edc.statemachine.retry.processor; | ||
|
|
||
| import org.eclipse.edc.spi.entity.StatefulEntity; | ||
|
|
||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.function.BiFunction; | ||
| import java.util.function.Function; | ||
|
|
||
| import static java.util.concurrent.CompletableFuture.failedFuture; | ||
|
|
||
| /** | ||
| * Process implementation that handles a process that returns {@link CompletableFuture} | ||
| * | ||
| * @param <E> entity type. | ||
| * @param <I> process input type. | ||
| * @param <O> process output type. | ||
| */ | ||
| public class FutureRetryProcess<E extends StatefulEntity<E>, I, O> implements Process<E, I, O> { | ||
|
|
||
| private final String name; | ||
| private final BiFunction<E, I, CompletableFuture<O>> function; | ||
| private Function<String, E> entityReload; | ||
|
|
||
| public FutureRetryProcess(String name, BiFunction<E, I, CompletableFuture<O>> function) { | ||
| this.name = name; | ||
| this.function = function; | ||
| } | ||
|
|
||
| @Override | ||
| public CompletableFuture<ProcessContext<E, O>> execute(ProcessContext<E, I> context) { | ||
| try { | ||
| return function.apply(context.entity(), context.content()) | ||
| .handle((content, throwable) -> { | ||
| var reloadedEntity = reloadEntity(context.entity()); | ||
| if (throwable == null) { | ||
| return new ProcessContext<>(reloadedEntity, content); | ||
| } else { | ||
| throw new EntityStateException(reloadedEntity, name, throwable.getMessage()); | ||
| } | ||
| }); | ||
| } catch (Throwable throwable) { | ||
| return failedFuture(new UnrecoverableEntityStateException(reloadEntity(context.entity()), name, throwable.getMessage())); | ||
| } | ||
| } | ||
|
|
||
| public FutureRetryProcess<E, I, O> entityReload(Function<String, E> entityReload) { | ||
| this.entityReload = entityReload; | ||
| return this; | ||
| } | ||
|
|
||
| private E reloadEntity(E entity) { | ||
| return entityReload == null ? entity : entityReload.apply(entity.getId()); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.