Performance optimization
This commit is contained in:
parent
c1747dc6dc
commit
8522e7a866
@ -19,6 +19,8 @@
|
|||||||
package org.warp.filesponge;
|
package org.warp.filesponge;
|
||||||
|
|
||||||
import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
|
import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
import java.util.concurrent.atomic.AtomicBoolean;
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
@ -74,28 +76,38 @@ public class FileSponge implements URLsHandler {
|
|||||||
@Override
|
@Override
|
||||||
public Flux<DataBlock> requestContent(URL url) {
|
public Flux<DataBlock> requestContent(URL url) {
|
||||||
AtomicBoolean alreadyPrintedDebug = new AtomicBoolean(false);
|
AtomicBoolean alreadyPrintedDebug = new AtomicBoolean(false);
|
||||||
return Flux
|
return Mono
|
||||||
.defer(() -> Flux.fromIterable(cacheAccess))
|
.fromCallable(() -> {
|
||||||
.map(urlsHandler -> urlsHandler.requestContent(url))
|
List<Flux<DataBlock>> contentRequests = new ArrayList<>(cacheAccess.size());
|
||||||
.collectList()
|
for (URLsDiskHandler urlsHandler : cacheAccess) {
|
||||||
|
contentRequests.add(urlsHandler.requestContent(url));
|
||||||
|
}
|
||||||
|
return contentRequests;
|
||||||
|
})
|
||||||
.flatMapMany(FileSpongeUtils::firstWithValueFlux)
|
.flatMapMany(FileSpongeUtils::firstWithValueFlux)
|
||||||
.doOnNext(dataBlock -> {
|
.doOnNext(dataBlock -> {
|
||||||
if (alreadyPrintedDebug.compareAndSet(false, true)) {
|
if (alreadyPrintedDebug.compareAndSet(false, true)) {
|
||||||
logger.debug("File \"{}\" content has been found in the cache", url);
|
logger.debug("File \"{}\" content has been found in the cache", url);
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
.switchIfEmpty(Flux
|
.switchIfEmpty(Mono
|
||||||
.defer(() -> Flux.fromIterable(urlsHandlers))
|
.fromCallable(() -> {
|
||||||
.doOnSubscribe(s -> logger.debug("Downloading file \"{}\" content", url))
|
logger.debug("Downloading file \"{}\" content", url);
|
||||||
.map(urlsHandler -> urlsHandler
|
List<Flux<DataBlock>> contentRequestsAndCaching = new ArrayList<>(urlsHandlers.size());
|
||||||
|
for (URLsHandler urlsHandler : urlsHandlers) {
|
||||||
|
contentRequestsAndCaching.add(urlsHandler
|
||||||
.requestContent(url)
|
.requestContent(url)
|
||||||
.flatMapSequential(dataBlock -> Flux
|
.flatMapSequential(dataBlock -> {
|
||||||
.defer(() -> Flux.fromIterable(cacheWrite))
|
List<Mono<Void>> cacheWriteActions = new ArrayList<>(cacheWrite.size());
|
||||||
.flatMapSequential(cw -> cw.writeContentBlock(url, dataBlock))
|
for (URLsWriter urlsWriter : cacheWrite) {
|
||||||
.then(Mono.just(dataBlock))
|
cacheWriteActions.add(urlsWriter.writeContentBlock(url, dataBlock));
|
||||||
)
|
}
|
||||||
)
|
return Mono.when(cacheWriteActions).thenReturn(dataBlock);
|
||||||
.collectList()
|
})
|
||||||
|
);
|
||||||
|
}
|
||||||
|
return contentRequestsAndCaching;
|
||||||
|
})
|
||||||
.flatMapMany(FileSpongeUtils::firstWithValueFlux)
|
.flatMapMany(FileSpongeUtils::firstWithValueFlux)
|
||||||
.doOnComplete(() -> logger.debug("Downloaded file \"{}\" content", url))
|
.doOnComplete(() -> logger.debug("Downloaded file \"{}\" content", url))
|
||||||
)
|
)
|
||||||
@ -106,28 +118,38 @@ public class FileSponge implements URLsHandler {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Mono<Metadata> requestMetadata(URL url) {
|
public Mono<Metadata> requestMetadata(URL url) {
|
||||||
return Flux
|
return Mono
|
||||||
.defer(() -> Flux.fromIterable(cacheAccess))
|
.fromCallable(() -> {
|
||||||
.map(urlsHandler -> urlsHandler.requestMetadata(url))
|
List<Mono<Metadata>> metadataRequests = new ArrayList<>(cacheAccess.size());
|
||||||
.collectList()
|
for (URLsDiskHandler urlsHandler : cacheAccess) {
|
||||||
|
metadataRequests.add(urlsHandler.requestMetadata(url));
|
||||||
|
}
|
||||||
|
return metadataRequests;
|
||||||
|
})
|
||||||
.flatMap(FileSpongeUtils::firstWithValueMono)
|
.flatMap(FileSpongeUtils::firstWithValueMono)
|
||||||
.doOnSuccess(metadata -> {
|
.doOnSuccess(metadata -> {
|
||||||
if (metadata != null) {
|
if (metadata != null) {
|
||||||
logger.debug("File \"{}\" metadata has been found in the cache", url);
|
logger.debug("File \"{}\" metadata has been found in the cache", url);
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
.switchIfEmpty(Flux
|
.switchIfEmpty(Mono
|
||||||
.defer(() -> Flux.fromIterable(urlsHandlers))
|
.fromCallable(() -> {
|
||||||
.doOnSubscribe(s -> logger.debug("Downloading file \"{}\" metadata", url))
|
logger.debug("Downloading file \"{}\" metadata", url);
|
||||||
.map(urlsHandler -> urlsHandler
|
List<Mono<Metadata>> metadataRequestsAndCaching = new ArrayList<>(urlsHandlers.size());
|
||||||
|
for (URLsHandler urlsHandler : urlsHandlers) {
|
||||||
|
metadataRequestsAndCaching.add(urlsHandler
|
||||||
.requestMetadata(url)
|
.requestMetadata(url)
|
||||||
.flatMap(dataBlock -> Flux
|
.flatMap(meta -> {
|
||||||
.defer(() -> Flux.fromIterable(cacheWrite))
|
List<Mono<Void>> cacheWriteActions = new ArrayList<>(cacheWrite.size());
|
||||||
.flatMapSequential(cw -> cw.writeMetadata(url, dataBlock))
|
for (URLsWriter urlsWriter : cacheWrite) {
|
||||||
.then(Mono.just(dataBlock))
|
cacheWriteActions.add(urlsWriter.writeMetadata(url, meta));
|
||||||
)
|
}
|
||||||
)
|
return Mono.when(cacheWriteActions).thenReturn(meta);
|
||||||
.collectList()
|
})
|
||||||
|
);
|
||||||
|
}
|
||||||
|
return metadataRequestsAndCaching;
|
||||||
|
})
|
||||||
.flatMap(FileSpongeUtils::firstWithValueMono)
|
.flatMap(FileSpongeUtils::firstWithValueMono)
|
||||||
.doOnSuccess(s -> {
|
.doOnSuccess(s -> {
|
||||||
if (s != null) {
|
if (s != null) {
|
||||||
|
Loading…
Reference in New Issue
Block a user