CavalliumDBEngine/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleMapped.java

201 lines
5.7 KiB
Java
Raw Normal View History

2021-01-31 19:52:47 +01:00
package it.cavallium.dbengine.database.collections;
2021-09-23 20:57:28 +02:00
import io.net5.buffer.api.Drop;
import io.net5.buffer.api.Owned;
import io.net5.buffer.api.Send;
import io.net5.buffer.api.internal.ResourceSupport;
2021-06-26 02:35:33 +02:00
import it.cavallium.dbengine.client.BadBlock;
2021-01-31 19:52:47 +01:00
import it.cavallium.dbengine.client.CompositeSnapshot;
2021-09-02 17:15:40 +02:00
import it.cavallium.dbengine.client.Mapper;
2021-05-08 03:09:00 +02:00
import it.cavallium.dbengine.database.Delta;
2021-09-24 01:59:56 +02:00
import it.cavallium.dbengine.database.LLEntry;
2021-05-08 03:09:00 +02:00
import it.cavallium.dbengine.database.LLUtils;
import it.cavallium.dbengine.database.UpdateReturnMode;
2021-08-22 21:23:22 +02:00
import it.cavallium.dbengine.database.serialization.SerializationException;
import it.cavallium.dbengine.database.serialization.SerializationFunction;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
2021-01-31 19:52:47 +01:00
import org.jetbrains.annotations.Nullable;
2021-06-26 02:35:33 +02:00
import reactor.core.publisher.Flux;
2021-01-31 19:52:47 +01:00
import reactor.core.publisher.Mono;
2021-08-22 21:23:22 +02:00
import reactor.core.publisher.SynchronousSink;
2021-01-31 19:52:47 +01:00
2021-04-12 17:09:55 +02:00
@SuppressWarnings("unused")
2021-09-23 20:57:28 +02:00
public class DatabaseSingleMapped<A, B> extends ResourceSupport<DatabaseStage<A>, DatabaseSingleMapped<A, B>>
implements DatabaseStageEntry<A> {
2021-01-31 19:52:47 +01:00
private static final Logger logger = LogManager.getLogger(DatabaseSingleMapped.class);
2021-10-01 19:17:33 +02:00
private static final Drop<DatabaseSingleMapped<?, ?>> DROP = new Drop<>() {
@Override
public void drop(DatabaseSingleMapped<?, ?> obj) {
try {
if (obj.serializedSingle != null) {
obj.serializedSingle.close();
}
} catch (Throwable ex) {
logger.error("Failed to close serializedSingle", ex);
}
}
@Override
public Drop<DatabaseSingleMapped<?, ?>> fork() {
return this;
}
@Override
public void attach(DatabaseSingleMapped<?, ?> obj) {
}
};
2021-09-02 17:15:40 +02:00
private final Mapper<A, B> mapper;
2021-01-31 19:52:47 +01:00
2021-09-23 20:57:28 +02:00
private DatabaseStageEntry<B> serializedSingle;
2021-10-01 19:17:33 +02:00
@SuppressWarnings({"unchecked", "rawtypes"})
2021-09-23 20:57:28 +02:00
public DatabaseSingleMapped(DatabaseStageEntry<B> serializedSingle, Mapper<A, B> mapper,
Drop<DatabaseSingleMapped<A, B>> drop) {
2021-10-01 19:17:33 +02:00
super((Drop<DatabaseSingleMapped<A,B>>) (Drop) DROP);
2021-01-31 19:52:47 +01:00
this.serializedSingle = serializedSingle;
2021-09-02 17:15:40 +02:00
this.mapper = mapper;
2021-01-31 19:52:47 +01:00
}
2021-10-01 19:17:33 +02:00
@SuppressWarnings({"unchecked", "rawtypes"})
2021-09-23 20:57:28 +02:00
private DatabaseSingleMapped(Send<DatabaseStage<B>> serializedSingle, Mapper<A, B> mapper,
Drop<DatabaseSingleMapped<A, B>> drop) {
2021-10-01 19:17:33 +02:00
super((Drop<DatabaseSingleMapped<A,B>>) (Drop) DROP);
2021-09-23 20:57:28 +02:00
this.mapper = mapper;
this.serializedSingle = (DatabaseStageEntry<B>) serializedSingle.receive();
}
2021-08-22 21:23:22 +02:00
private void deserializeSink(B value, SynchronousSink<A> sink) {
try {
2021-09-02 17:15:40 +02:00
sink.next(this.unMap(value));
2021-08-22 21:23:22 +02:00
} catch (SerializationException ex) {
sink.error(ex);
}
}
2021-01-31 19:52:47 +01:00
@Override
2021-04-12 17:09:55 +02:00
public Mono<A> get(@Nullable CompositeSnapshot snapshot, boolean existsAlmostCertainly) {
2021-08-22 21:23:22 +02:00
return serializedSingle.get(snapshot, existsAlmostCertainly).handle(this::deserializeSink);
2021-01-31 19:52:47 +01:00
}
@Override
2021-04-12 17:09:55 +02:00
public Mono<A> getOrDefault(@Nullable CompositeSnapshot snapshot, Mono<A> defaultValue) {
2021-08-22 21:23:22 +02:00
return serializedSingle.get(snapshot).handle(this::deserializeSink).switchIfEmpty(defaultValue);
2021-01-31 19:52:47 +01:00
}
@Override
2021-04-12 17:09:55 +02:00
public Mono<Void> set(A value) {
2021-08-22 21:23:22 +02:00
return Mono
2021-09-02 17:15:40 +02:00
.fromCallable(() -> map(value))
2021-08-22 21:23:22 +02:00
.flatMap(serializedSingle::set);
2021-01-31 19:52:47 +01:00
}
@Override
2021-04-12 17:09:55 +02:00
public Mono<A> setAndGetPrevious(A value) {
2021-08-22 21:23:22 +02:00
return Mono
2021-09-02 17:15:40 +02:00
.fromCallable(() -> map(value))
2021-08-22 21:23:22 +02:00
.flatMap(serializedSingle::setAndGetPrevious)
.handle(this::deserializeSink);
2021-01-31 19:52:47 +01:00
}
@Override
public Mono<Boolean> setAndGetChanged(A value) {
2021-08-22 21:23:22 +02:00
return Mono
2021-09-02 17:15:40 +02:00
.fromCallable(() -> map(value))
2021-08-22 21:23:22 +02:00
.flatMap(serializedSingle::setAndGetChanged)
.single();
2021-01-31 19:52:47 +01:00
}
2021-02-06 19:21:31 +01:00
@Override
2021-08-22 21:23:22 +02:00
public Mono<A> update(SerializationFunction<@Nullable A, @Nullable A> updater,
2021-05-08 03:09:00 +02:00
UpdateReturnMode updateReturnMode,
boolean existsAlmostCertainly) {
return serializedSingle.update(oldValue -> {
2021-09-02 17:15:40 +02:00
var result = updater.apply(oldValue == null ? null : this.unMap(oldValue));
if (result == null) {
return null;
} else {
2021-09-02 17:15:40 +02:00
return this.map(result);
}
2021-08-22 21:23:22 +02:00
}, updateReturnMode, existsAlmostCertainly).handle(this::deserializeSink);
2021-05-08 03:09:00 +02:00
}
@Override
2021-08-22 21:23:22 +02:00
public Mono<Delta<A>> updateAndGetDelta(SerializationFunction<@Nullable A, @Nullable A> updater,
2021-05-08 03:09:00 +02:00
boolean existsAlmostCertainly) {
return serializedSingle.updateAndGetDelta(oldValue -> {
2021-09-02 17:15:40 +02:00
var result = updater.apply(oldValue == null ? null : this.unMap(oldValue));
2021-05-08 03:09:00 +02:00
if (result == null) {
return null;
} else {
2021-09-02 17:15:40 +02:00
return this.map(result);
2021-05-08 03:09:00 +02:00
}
2021-09-02 17:15:40 +02:00
}, existsAlmostCertainly).transform(mono -> LLUtils.mapDelta(mono, this::unMap));
2021-02-06 19:21:31 +01:00
}
2021-01-31 19:52:47 +01:00
@Override
public Mono<Void> clear() {
return serializedSingle.clear();
}
@Override
2021-04-12 17:09:55 +02:00
public Mono<A> clearAndGetPrevious() {
2021-08-22 21:23:22 +02:00
return serializedSingle.clearAndGetPrevious().handle(this::deserializeSink);
2021-01-31 19:52:47 +01:00
}
@Override
public Mono<Boolean> clearAndGetStatus() {
return serializedSingle.clearAndGetStatus();
}
@Override
2021-02-24 16:43:07 +01:00
public Mono<Long> leavesCount(@Nullable CompositeSnapshot snapshot, boolean fast) {
return serializedSingle.leavesCount(snapshot, fast);
2021-01-31 19:52:47 +01:00
}
@Override
public Mono<Boolean> isEmpty(@Nullable CompositeSnapshot snapshot) {
return serializedSingle.isEmpty(snapshot);
}
@Override
2021-04-12 17:09:55 +02:00
public DatabaseStageEntry<A> entry() {
2021-01-31 19:52:47 +01:00
return this;
}
2021-06-26 02:35:33 +02:00
@Override
public Flux<BadBlock> badBlocks() {
return this.serializedSingle.badBlocks();
}
2021-09-02 17:15:40 +02:00
private A unMap(B bytes) throws SerializationException {
return mapper.unmap(bytes);
2021-01-31 19:52:47 +01:00
}
2021-09-02 17:15:40 +02:00
private B map(A bytes) throws SerializationException {
return mapper.map(bytes);
2021-01-31 19:52:47 +01:00
}
2021-09-23 20:57:28 +02:00
@Override
protected RuntimeException createResourceClosedException() {
throw new IllegalStateException("Closed");
}
@Override
protected Owned<DatabaseSingleMapped<A, B>> prepareSend() {
var serializedSingle = this.serializedSingle.send();
return drop -> new DatabaseSingleMapped<>(serializedSingle, mapper, drop);
}
@Override
protected void makeInaccessible() {
this.serializedSingle = null;
}
2021-01-31 19:52:47 +01:00
}