andreacavalli/FileSponge
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Copy on write to avoid errors

Browse Source
This commit is contained in:
Andrea Cavalli 2021-06-29 23:49:52 +02:00
parent daf48f0330
commit a8aa1bb98d
1 changed files with 38 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
src/main/java/org/warp/filesponge/FileSponge.java
View File

@ -18,6 +18,7 @@
package org.warp.filesponge; package org.warp.filesponge;
import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
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;
@ -33,89 +34,101 @@ public class FileSponge implements URLsHandler {
public static final int BLOCK_SIZE = 1024 * 1024; // 1 MiB public static final int BLOCK_SIZE = 1024 * 1024; // 1 MiB
private final Set<URLsHandler> urlsHandlers = new ConcurrentHashMap<URLsHandler, Object>().keySet(new Object()); private final Object structuresLock = new Object();
private volatile ObjectOpenHashSet<URLsHandler> urlsHandlers = ObjectOpenHashSet.of();
private final Set<URLsDiskHandler> cacheAccess = new ConcurrentHashMap<URLsDiskHandler, Object>().keySet(new Object()); private volatile ObjectOpenHashSet<URLsDiskHandler> cacheAccess = ObjectOpenHashSet.of();
private final Set<URLsWriter> cacheWrite = new ConcurrentHashMap<URLsWriter, Object>().keySet(new Object()); private volatile ObjectOpenHashSet<URLsWriter> cacheWrite = ObjectOpenHashSet.of();
public FileSponge() { public FileSponge() {
} }
public Mono<Void> registerSource(URLsHandler urLsHandler) { public Mono<Void> registerSource(URLsHandler urLsHandler) {
return Mono.fromRunnable(() -> urlsHandlers.add(urLsHandler)); return Mono
.<Void>fromRunnable(() -> {
synchronized (structuresLock) {
var clone = urlsHandlers.clone();
clone.add(urLsHandler);
this.urlsHandlers = clone;
}
})
.subscribeOn(Schedulers.boundedElastic());
} }
public <T extends URLsDiskHandler & URLsWriter> Mono<Void> registerCache(T urlsCache) { public <T extends URLsDiskHandler & URLsWriter> Mono<Void> registerCache(T urlsCache) {
return Mono.fromRunnable(() -> { return Mono
cacheAccess.add(urlsCache); .<Void>fromRunnable(() -> {
cacheWrite.add(urlsCache); synchronized (structuresLock) {
}); var cacheAccessClone = cacheAccess.clone();
cacheAccessClone.add(urlsCache);
this.cacheAccess = cacheAccessClone;
var cacheWriteClone = cacheWrite.clone();
cacheWriteClone.add(urlsCache);
this.cacheWrite = cacheWriteClone;
}
})
.subscribeOn(Schedulers.boundedElastic());
} }
@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 Flux
.fromStream(cacheAccess::stream) .defer(() -> Flux.fromIterable(cacheAccess))
.map(urlsHandler -> urlsHandler.requestContent(url)) .map(urlsHandler -> urlsHandler.requestContent(url))
.collectList() .collectList()
.flatMapMany(monos -> FileSpongeUtils.firstWithValueFlux(monos)) .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(Flux
.fromStream(urlsHandlers::stream) .defer(() -> Flux.fromIterable(urlsHandlers))
.doOnSubscribe(s -> logger.debug("Downloading file \"{}\" content", url)) .doOnSubscribe(s -> logger.debug("Downloading file \"{}\" content", url))
.map(urlsHandler -> urlsHandler .map(urlsHandler -> urlsHandler
.requestContent(url) .requestContent(url)
.flatMapSequential(dataBlock -> Flux .flatMapSequential(dataBlock -> Flux
.fromStream(cacheWrite::stream) .defer(() -> Flux.fromIterable(cacheWrite))
.flatMapSequential(cw -> cw.writeContentBlock(url, dataBlock)) .flatMapSequential(cw -> cw.writeContentBlock(url, dataBlock))
.then(Mono.just(dataBlock)) .then(Mono.just(dataBlock))
) )
) )
.collectList() .collectList()
.flatMapMany(monos -> FileSpongeUtils.firstWithValueFlux(monos)) .flatMapMany(FileSpongeUtils::firstWithValueFlux)
.doOnComplete(() -> logger.debug("Downloaded file \"{}\" content", url)) .doOnComplete(() -> logger.debug("Downloaded file \"{}\" content", url))
) )
.distinct(DataBlock::getId) .distinct(DataBlock::getId)
.doOnDiscard(DataBlock.class, DataBlock::release) .doOnDiscard(DataBlock.class, DataBlock::release);
.doOnDiscard(Flux.class, f -> {
//noinspection unchecked
Flux<DataBlock> flux = (Flux<DataBlock>) f;
flux.doOnNext(DataBlock::release).subscribeOn(Schedulers.single()).subscribe();
});
} }
@Override @Override
public Mono<Metadata> requestMetadata(URL url) { public Mono<Metadata> requestMetadata(URL url) {
return Flux return Flux
.fromStream(cacheAccess::stream) .defer(() -> Flux.fromIterable(cacheAccess))
.map(urlsHandler -> urlsHandler.requestMetadata(url)) .map(urlsHandler -> urlsHandler.requestMetadata(url))
.collectList() .collectList()
.flatMap(monos -> FileSpongeUtils.firstWithValueMono(monos)) .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(Flux
.fromStream(urlsHandlers::stream) .defer(() -> Flux.fromIterable(urlsHandlers))
.doOnSubscribe(s -> logger.debug("Downloading file \"{}\" metadata", url)) .doOnSubscribe(s -> logger.debug("Downloading file \"{}\" metadata", url))
.map(urlsHandler -> urlsHandler .map(urlsHandler -> urlsHandler
.requestMetadata(url) .requestMetadata(url)
.flatMap(dataBlock -> Flux .flatMap(dataBlock -> Flux
.fromStream(cacheWrite::stream) .defer(() -> Flux.fromIterable(cacheWrite))
.flatMapSequential(cw -> cw.writeMetadata(url, dataBlock)) .flatMapSequential(cw -> cw.writeMetadata(url, dataBlock))
.then(Mono.just(dataBlock)) .then(Mono.just(dataBlock))
) )
) )
.collectList() .collectList()
.flatMap(monos -> FileSpongeUtils.firstWithValueMono(monos)) .flatMap(FileSpongeUtils::firstWithValueMono)
.doOnSuccess(s -> { .doOnSuccess(s -> {
if (s != null) { if (s != null) {
logger.debug("Downloaded file \"{}\" metadata", url); logger.debug("Downloaded file \"{}\" metadata", url);