Skip to content

Commit

Permalink
Merge branch 'develop'
Browse files Browse the repository at this point in the history
  • Loading branch information
GangCheng committed Aug 18, 2023
2 parents b4c5387 + 5e6884e commit c9933ca
Show file tree
Hide file tree
Showing 9 changed files with 329 additions and 251 deletions.
2 changes: 1 addition & 1 deletion .github/badges/jacoco.svg
View file
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
168 changes: 70 additions & 98 deletions .../chenggang/project/reactive/cache/support/defaults/executor/DefaultReactiveFluxCache.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;

/**
* The default reactive flux cache
Expand Down Expand Up @@ -75,8 +74,6 @@ public <T> Flux<T> cacheIfNecessary(@NonNull String cacheKey,
);
return reactiveCacheFluxAdapter.loadData(cacheKey);
}
// initialization is idle
final AtomicBoolean alreadyReleaseFlag = new AtomicBoolean(false);
return Flux.usingWhen(
this.reactiveCacheLock.tryLockInitializeLock(cacheName,
cacheKey,
Expand All @@ -94,60 +91,48 @@ public <T> Flux<T> cacheIfNecessary(@NonNull String cacheKey,
))
)
,
currentOperationId -> {
if (alreadyReleaseFlag.compareAndSet(false, true)) {
return this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Flux)Release initialization lock, CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then();
}
return Flux.empty();
},
(currentOperationId, throwable) -> {
if (alreadyReleaseFlag.compareAndSet(false, true)) {
return this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Flux)Release initialization lock on Error, " +
"CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then();
}
return Flux.empty();
},
(currentOperationId) -> {
if (alreadyReleaseFlag.compareAndSet(false, true)) {
return this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Flux)Release initialization lock, " +
"CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then();
}
return Flux.empty();
}
currentOperationId -> this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Flux)Release initialization lock, CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then()
,
(currentOperationId, throwable) -> this.reactiveCacheLock.releaseInitializeLock(cacheName,
cacheKey
)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Flux)Release initialization lock on Error, " +
"CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then()
,
(currentOperationId) -> this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Flux)Release initialization lock, " +
"CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then()
);
});
}

@Override
public Mono<Void> evictCache(@NonNull String cacheKey) {
final AtomicBoolean alreadyReleaseFlag = new AtomicBoolean(false);
return Mono.usingWhen(
this.reactiveCacheLock.tryLockInitializeLock(cacheName,
cacheKey,
Expand All @@ -157,53 +142,40 @@ public Mono<Void> evictCache(@NonNull String cacheKey) {
.filter(Boolean::booleanValue)
.flatMap(hasData -> reactiveCacheFluxAdapter.cleanupData(cacheKey))
,
currentOperationId -> {
if (alreadyReleaseFlag.compareAndSet(false, true)) {
return this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Cleanup)Release initialization lock, CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then();
}
return Mono.empty();
},
(currentOperationId, throwable) -> {
if (alreadyReleaseFlag.compareAndSet(false, true)) {
return this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Cleanup)Release initialization lock on Error, " +
"CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then(Mono.error(throwable));
}
return Mono.error(throwable);
},
(currentOperationId) -> {
if (alreadyReleaseFlag.compareAndSet(false, true)) {
return this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Cleanup)Release initialization lock, " +
"CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then();
}
return Mono.empty();
}
currentOperationId -> this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Cleanup)Release initialization lock, CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then()
,
(currentOperationId, throwable) -> this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Cleanup)Release initialization lock on Error, " +
"CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then()
,
(currentOperationId) -> this.reactiveCacheLock.releaseInitializeLock(cacheName, cacheKey)
.doOnNext(operationId -> log.debug(
"[Reactive Cache](Cleanup)Release initialization lock, " +
"CacheName:{}, CacheKey:{}, " +
"ReleasedOperationId:{}, CurrentOperationId:{}",
cacheName,
cacheKey,
operationId,
currentOperationId
))
.then()
);
}
}
Loading

0 comments on commit c9933ca

Please sign in to comment.