Atomic access to data structures
This commit is contained in:
parent
8522e7a866
commit
7d8c236a8f
@ -61,7 +61,7 @@ public class FileSponge implements URLsHandler {
|
|||||||
return Mono
|
return Mono
|
||||||
.<Void>fromRunnable(() -> {
|
.<Void>fromRunnable(() -> {
|
||||||
synchronized (structuresLock) {
|
synchronized (structuresLock) {
|
||||||
var cacheAccessClone = cacheAccess.clone();
|
var cacheAccessClone = this.cacheAccess.clone();
|
||||||
cacheAccessClone.add(urlsCache);
|
cacheAccessClone.add(urlsCache);
|
||||||
this.cacheAccess = cacheAccessClone;
|
this.cacheAccess = cacheAccessClone;
|
||||||
|
|
||||||
@ -78,8 +78,9 @@ public class FileSponge implements URLsHandler {
|
|||||||
AtomicBoolean alreadyPrintedDebug = new AtomicBoolean(false);
|
AtomicBoolean alreadyPrintedDebug = new AtomicBoolean(false);
|
||||||
return Mono
|
return Mono
|
||||||
.fromCallable(() -> {
|
.fromCallable(() -> {
|
||||||
List<Flux<DataBlock>> contentRequests = new ArrayList<>(cacheAccess.size());
|
var ca = this.cacheAccess;
|
||||||
for (URLsDiskHandler urlsHandler : cacheAccess) {
|
List<Flux<DataBlock>> contentRequests = new ArrayList<>(ca.size());
|
||||||
|
for (URLsDiskHandler urlsHandler : ca) {
|
||||||
contentRequests.add(urlsHandler.requestContent(url));
|
contentRequests.add(urlsHandler.requestContent(url));
|
||||||
}
|
}
|
||||||
return contentRequests;
|
return contentRequests;
|
||||||
@ -93,13 +94,15 @@ public class FileSponge implements URLsHandler {
|
|||||||
.switchIfEmpty(Mono
|
.switchIfEmpty(Mono
|
||||||
.fromCallable(() -> {
|
.fromCallable(() -> {
|
||||||
logger.debug("Downloading file \"{}\" content", url);
|
logger.debug("Downloading file \"{}\" content", url);
|
||||||
List<Flux<DataBlock>> contentRequestsAndCaching = new ArrayList<>(urlsHandlers.size());
|
var uh = this.urlsHandlers;
|
||||||
for (URLsHandler urlsHandler : urlsHandlers) {
|
List<Flux<DataBlock>> contentRequestsAndCaching = new ArrayList<>(uh.size());
|
||||||
|
for (URLsHandler urlsHandler : uh) {
|
||||||
contentRequestsAndCaching.add(urlsHandler
|
contentRequestsAndCaching.add(urlsHandler
|
||||||
.requestContent(url)
|
.requestContent(url)
|
||||||
.flatMapSequential(dataBlock -> {
|
.flatMapSequential(dataBlock -> {
|
||||||
List<Mono<Void>> cacheWriteActions = new ArrayList<>(cacheWrite.size());
|
var cw = this.cacheWrite;
|
||||||
for (URLsWriter urlsWriter : cacheWrite) {
|
List<Mono<Void>> cacheWriteActions = new ArrayList<>(cw.size());
|
||||||
|
for (URLsWriter urlsWriter : cw) {
|
||||||
cacheWriteActions.add(urlsWriter.writeContentBlock(url, dataBlock));
|
cacheWriteActions.add(urlsWriter.writeContentBlock(url, dataBlock));
|
||||||
}
|
}
|
||||||
return Mono.when(cacheWriteActions).thenReturn(dataBlock);
|
return Mono.when(cacheWriteActions).thenReturn(dataBlock);
|
||||||
@ -120,8 +123,9 @@ public class FileSponge implements URLsHandler {
|
|||||||
public Mono<Metadata> requestMetadata(URL url) {
|
public Mono<Metadata> requestMetadata(URL url) {
|
||||||
return Mono
|
return Mono
|
||||||
.fromCallable(() -> {
|
.fromCallable(() -> {
|
||||||
List<Mono<Metadata>> metadataRequests = new ArrayList<>(cacheAccess.size());
|
var ca = this.cacheAccess;
|
||||||
for (URLsDiskHandler urlsHandler : cacheAccess) {
|
List<Mono<Metadata>> metadataRequests = new ArrayList<>(ca.size());
|
||||||
|
for (URLsDiskHandler urlsHandler : ca) {
|
||||||
metadataRequests.add(urlsHandler.requestMetadata(url));
|
metadataRequests.add(urlsHandler.requestMetadata(url));
|
||||||
}
|
}
|
||||||
return metadataRequests;
|
return metadataRequests;
|
||||||
@ -135,13 +139,15 @@ public class FileSponge implements URLsHandler {
|
|||||||
.switchIfEmpty(Mono
|
.switchIfEmpty(Mono
|
||||||
.fromCallable(() -> {
|
.fromCallable(() -> {
|
||||||
logger.debug("Downloading file \"{}\" metadata", url);
|
logger.debug("Downloading file \"{}\" metadata", url);
|
||||||
List<Mono<Metadata>> metadataRequestsAndCaching = new ArrayList<>(urlsHandlers.size());
|
var uh = this.urlsHandlers;
|
||||||
for (URLsHandler urlsHandler : urlsHandlers) {
|
List<Mono<Metadata>> metadataRequestsAndCaching = new ArrayList<>(uh.size());
|
||||||
|
for (URLsHandler urlsHandler : uh) {
|
||||||
metadataRequestsAndCaching.add(urlsHandler
|
metadataRequestsAndCaching.add(urlsHandler
|
||||||
.requestMetadata(url)
|
.requestMetadata(url)
|
||||||
.flatMap(meta -> {
|
.flatMap(meta -> {
|
||||||
List<Mono<Void>> cacheWriteActions = new ArrayList<>(cacheWrite.size());
|
var cw = this.cacheWrite;
|
||||||
for (URLsWriter urlsWriter : cacheWrite) {
|
List<Mono<Void>> cacheWriteActions = new ArrayList<>(cw.size());
|
||||||
|
for (URLsWriter urlsWriter : cw) {
|
||||||
cacheWriteActions.add(urlsWriter.writeMetadata(url, meta));
|
cacheWriteActions.add(urlsWriter.writeMetadata(url, meta));
|
||||||
}
|
}
|
||||||
return Mono.when(cacheWriteActions).thenReturn(meta);
|
return Mono.when(cacheWriteActions).thenReturn(meta);
|
||||||
|
Loading…
Reference in New Issue
Block a user