Skip to content

Commit

Permalink
feat: renamed StateRollback to StateRevert (#2)
Browse files Browse the repository at this point in the history
  • Loading branch information
@Dennis-Koch
Dennis-Koch authored Jun 24, 2024
1 parent 4c8dcb1 commit 8feb8fe
Show file tree
Hide file tree
Showing 37 changed files with 728 additions and 744 deletions.
75 changes: 30 additions & 45 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,12 +41,10 @@ So lets try a better version - still without dedicated library support...

```java
setFooState(fooValue1);
try{

run_logic_that_works_with_state_of_foo();
}finally{

clearFooState();
try {
run_logic_that_works_with_state_of_foo();
} finally {
clearFooState();
}
```

Expand Down Expand Up @@ -74,23 +72,18 @@ addition to our foo state. And we make it so that the fooState is applied condit
```java
var oldFooState = getFooState();
var oldBarState = getBarState();
if(conditionIsSatisfied){

setFooState(fooValue1);
if (conditionIsSatisfied) {
setFooState(fooValue1);
}

setBarState(barValue1);
try{

run_logic_that_works_with_state_of_foo_and_bar();
}finally{

setBarState(oldBarState);
if(conditionIsSatisfied){

setFooState(oldFooState);
try {
run_logic_that_works_with_state_of_foo_and_bar();
} finally {
setBarState(oldBarState);
if (conditionIsSatisfied) {
setFooState(oldFooState);
}
}
}
```

**You might already expect it: This approach - despite all the increased clutter that we had to produce already - still
Expand All @@ -105,27 +98,22 @@ our "transactional" problem. An ugly - but admitted truly robust - solution woul

```java
var oldFooState = getFooState();
if(conditionIsSatisfied){

setFooState(fooValue1);
if (conditionIsSatisfied) {
setFooState(fooValue1);
}
try{
var oldBarState = getBarState();

setBarState(barValue1);
try{

run_logic_that_works_with_state_of_foo_and_bar();
}finally{

setBarState(oldBarState);
try {
var oldBarState = getBarState();
setBarState(barValue1);
try {
run_logic_that_works_with_state_of_foo_and_bar();
} finally {
setBarState(oldBarState);
}
}finally{
if(conditionIsSatisfied){

setFooState(oldFooState);
} finally {
if(conditionIsSatisfied) {
setFooState(oldFooState);
}
}
}
```

In real case scenarios it often becomes even more complex than what we have seen here in our example above. Normally the
Expand All @@ -143,13 +131,10 @@ var revert = DefaultStateRevert.chain(chain -> {
}
chain.append(pushBarState(barValue1));
});
try{

run_logic_that_works_with_state_of_foo_and_bar();
}finally{
revert.

revert();
try {
run_logic_that_works_with_state_of_foo_and_bar();
} finally {
revert.revert();
}
```

Expand Down
4 changes: 2 additions & 2 deletions threadly-streaming/src/main/java/org/threadlys/state/StateFetcherFactoryImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,11 +72,11 @@ public <T> Future<T> submit(Callable<T> task) {

protected <T> T executeCallableAndCleanupWorkerState(Callable<T> task) throws Exception {
var cs = contextSnapshotFactory.createSnapshot();
var rollback = cs.apply();
var revert = cs.apply();
try {
return task.call();
} finally {
rollback.rollback();
revert.revert();
}
}

Expand Down
14 changes: 7 additions & 7 deletions threadly-streaming/src/main/java/org/threadlys/streams/DataProcessorExtendable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,15 @@

import java.util.Collection;

import org.threadlys.utils.IStateRollback;
import org.threadlys.utils.StateRollback;
import org.threadlys.utils.StateRevert;
import org.threadlys.utils.DefaultStateRevert;

/**
* Defines extension points for registering data processors in order to make them available within the {@link AsyncDataProcessor} engine
*/
public interface DataProcessorExtendable {
default <E> IStateRollback registerDataProcessor(DataProcessor<E, ?> dataProcessor, Class<? extends E> entityType, Collection<DataScope> dataScopes, Collection<DataScope> requiredDataScopes) {
return StateRollback.chain(chain -> {
default <E> StateRevert registerDataProcessor(DataProcessor<E, ?> dataProcessor, Class<? extends E> entityType, Collection<DataScope> dataScopes, Collection<DataScope> requiredDataScopes) {
return DefaultStateRevert.chain(chain -> {
if (dataScopes != null) {
for (DataScope dataScope : dataScopes) {
chain.append(registerDataProcessor(dataProcessor, entityType, dataScope));
Expand All @@ -32,7 +32,7 @@ default <E> IStateRollback registerDataProcessor(DataProcessor<E, ?> dataProcess
* @param entityType
* @param dataScope
*/
<E> IStateRollback registerDataProcessor(DataProcessor<E, ?> dataProcessor, Class<? extends E> entityType, DataScope dataScope);
<E> StateRevert registerDataProcessor(DataProcessor<E, ?> dataProcessor, Class<? extends E> entityType, DataScope dataScope);

/**
* Registers a data processor to require the given data scope before executing this data processor. Normally the required data scope of maintained by another data processor. This way you can
Expand All @@ -42,7 +42,7 @@ default <E> IStateRollback registerDataProcessor(DataProcessor<E, ?> dataProcess
* @param dataProcessor
* @param requiredDataScope
*/
<E> IStateRollback registerDataProcessorDependency(DataProcessor<E, ?> dataProcessor, DataScope requiredDataScope);
<E> StateRevert registerDataProcessorDependency(DataProcessor<E, ?> dataProcessor, DataScope requiredDataScope);

/**
* Registers a data processor to require the given exception handler for unhandled exceptions. Executions of {@link DataProcessor#process(Object)} will be enclosed with a try/catch and exceptions
Expand All @@ -52,5 +52,5 @@ default <E> IStateRollback registerDataProcessor(DataProcessor<E, ?> dataProcess
* @param dataProcessor
* @param exceptionHandler
*/
<E> IStateRollback registerDataProcessorExceptionHandler(DataProcessor<E, ?> dataProcessor, DataProcessorExceptionHandler exceptionHandler);
<E> StateRevert registerDataProcessorExceptionHandler(DataProcessor<E, ?> dataProcessor, DataProcessorExceptionHandler exceptionHandler);
}
24 changes: 12 additions & 12 deletions threadly-streaming/src/main/java/org/threadlys/streams/impl/AsyncDataProcessorImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,10 @@
import org.threadlys.threading.ContextSnapshot;
import org.threadlys.threading.ContextSnapshotFactory;
import org.threadlys.threading.impl.ForkJoinPoolGuard;
import org.threadlys.utils.IStateRollback;
import org.threadlys.utils.StateRevert;
import org.threadlys.utils.ListenersMapListAdapter;
import org.threadlys.utils.SneakyThrowUtil;
import org.threadlys.utils.StateRollback;
import org.threadlys.utils.DefaultStateRevert;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

Expand Down Expand Up @@ -353,12 +353,12 @@ protected <E, C extends DataProcessorContext> void applyDataProcessorsToEntities
}

protected <E, C extends DataProcessorContext> void executeDataProcessorStages(List<List<RunnableSupplier<E, C>>> stageToRunnableSuppliersList) {
var rollback = StateRollback.empty();
var revert = DefaultStateRevert.empty();
try {
var fjp = forkJoinPoolGuard.currentForkJoinPool();
if (fjp == null) {
fjp = forkJoinPoolGuard.getDefaultForkJoinPool();
rollback = forkJoinPoolGuard.pushForkJoinPool(fjp);
revert = forkJoinPoolGuard.pushForkJoinPool(fjp);
}
for (var runnableSuppliersList : stageToRunnableSuppliersList) {
var callables = new ArrayList<Callable<CheckedConsumer<E>>>(runnableSuppliersList.size());
Expand All @@ -385,7 +385,7 @@ protected <E, C extends DataProcessorContext> void executeDataProcessorStages(Li
updateEntities(futures, indexToEntityMap, fjp);
}
} finally {
rollback.rollback();
revert.revert();
}
}

Expand Down Expand Up @@ -769,31 +769,31 @@ protected <E, C extends DataProcessorContext> Callable<CheckedConsumer<E>> async
var exceptionHandlers = dataProcessorToExceptionHandlerMap.get(dataProcessor);
if (exceptionHandlers == null || exceptionHandlers.isEmpty()) {
return () -> {
var rollback = cs.apply();
var revert = cs.apply();
try {
return dataProcessor.process(processorContext);
} finally {
rollback.rollback();
revert.revert();
}
};
} else {
return () -> {
var rollback = cs.apply();
var revert = cs.apply();
try {
return dataProcessor.process(processorContext);
} catch (Throwable e) {
var lastExceptionHandler = (DataProcessorExceptionHandler) exceptionHandlers.get(exceptionHandlers.size() - 1);
return lastExceptionHandler.handleProcessException(dataProcessor, processorContext, e);
} finally {
rollback.rollback();
revert.revert();
}
};
}
}

@SuppressWarnings({ "rawtypes", "unchecked" })
@Override
public <E> IStateRollback registerDataProcessor(DataProcessor<E, ?> dataProcessor, Class<? extends E> entityType, DataScope dataScope) {
public <E> StateRevert registerDataProcessor(DataProcessor<E, ?> dataProcessor, Class<? extends E> entityType, DataScope dataScope) {
writeLock.lock();
try {
ConfigurationState newState = new ConfigurationState(state);
Expand Down Expand Up @@ -831,7 +831,7 @@ protected <E> void unregisterDataProcessor(DataProcessor<E, ?> dataProcessor, Cl
}

@Override
public <E> IStateRollback registerDataProcessorDependency(DataProcessor<E, ?> dataProcessor, DataScope requiredDataScope) {
public <E> StateRevert registerDataProcessorDependency(DataProcessor<E, ?> dataProcessor, DataScope requiredDataScope) {
writeLock.lock();
try {
ConfigurationState newState = new ConfigurationState(state);
Expand All @@ -855,7 +855,7 @@ protected <E> void unregisterDataProcessorDependency(DataProcessor<E, ?> dataPro
}

@Override
public <E> IStateRollback registerDataProcessorExceptionHandler(DataProcessor<E, ?> dataProcessor, DataProcessorExceptionHandler exceptionHandler) {
public <E> StateRevert registerDataProcessorExceptionHandler(DataProcessor<E, ?> dataProcessor, DataProcessorExceptionHandler exceptionHandler) {
writeLock.lock();
try {
ConfigurationState newState = new ConfigurationState(state);
Expand Down
Loading

0 comments on commit 8feb8fe

Please sign in to comment.