2021-01-31 00:36:21 +01:00
|
|
|
package it.cavallium.dbengine.database.collections;
|
|
|
|
|
2021-09-17 16:56:28 +02:00
|
|
|
import io.net5.buffer.api.Buffer;
|
|
|
|
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 00:36:21 +01:00
|
|
|
import it.cavallium.dbengine.client.CompositeSnapshot;
|
2021-05-08 03:09:00 +02:00
|
|
|
import it.cavallium.dbengine.database.Delta;
|
2021-01-31 00:36:21 +01:00
|
|
|
import it.cavallium.dbengine.database.LLDictionary;
|
|
|
|
import it.cavallium.dbengine.database.LLDictionaryResultType;
|
|
|
|
import it.cavallium.dbengine.database.LLRange;
|
|
|
|
import it.cavallium.dbengine.database.LLSnapshot;
|
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;
|
2021-02-02 19:40:37 +01:00
|
|
|
import it.cavallium.dbengine.database.serialization.Serializer;
|
2021-01-31 00:36:21 +01:00
|
|
|
import org.jetbrains.annotations.Nullable;
|
2021-06-26 02:35:33 +02:00
|
|
|
import reactor.core.publisher.Flux;
|
2021-01-31 00:36:21 +01:00
|
|
|
import reactor.core.publisher.Mono;
|
2021-08-22 21:23:22 +02:00
|
|
|
import reactor.core.publisher.SynchronousSink;
|
2021-01-31 00:36:21 +01:00
|
|
|
|
2021-01-31 19:52:47 +01:00
|
|
|
public class DatabaseSingle<U> implements DatabaseStageEntry<U> {
|
2021-01-31 00:36:21 +01:00
|
|
|
|
|
|
|
private final LLDictionary dictionary;
|
2021-08-29 23:18:03 +02:00
|
|
|
private final Buffer key;
|
|
|
|
private final Mono<Send<Buffer>> keyMono;
|
2021-09-02 17:15:40 +02:00
|
|
|
private final Serializer<U> serializer;
|
2021-01-31 00:36:21 +01:00
|
|
|
|
2021-09-02 17:15:40 +02:00
|
|
|
public DatabaseSingle(LLDictionary dictionary, Send<Buffer> key, Serializer<U> serializer) {
|
2021-08-29 23:18:03 +02:00
|
|
|
try (key) {
|
2021-05-02 19:18:15 +02:00
|
|
|
this.dictionary = dictionary;
|
2021-08-29 23:18:03 +02:00
|
|
|
this.key = key.receive();
|
2021-08-22 18:20:05 +02:00
|
|
|
this.keyMono = LLUtils.lazyRetain(this.key);
|
2021-05-02 19:18:15 +02:00
|
|
|
this.serializer = serializer;
|
2021-04-30 19:15:04 +02:00
|
|
|
}
|
2021-01-31 00:36:21 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
private LLSnapshot resolveSnapshot(@Nullable CompositeSnapshot snapshot) {
|
|
|
|
if (snapshot == null) {
|
|
|
|
return null;
|
|
|
|
} else {
|
|
|
|
return snapshot.getSnapshot(dictionary);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-08-29 23:18:03 +02:00
|
|
|
private void deserializeValue(Send<Buffer> value, SynchronousSink<U> sink) {
|
2021-09-08 00:22:39 +02:00
|
|
|
try (value) {
|
2021-09-02 17:15:40 +02:00
|
|
|
sink.next(serializer.deserialize(value).deserializedData());
|
2021-08-22 21:23:22 +02:00
|
|
|
} catch (SerializationException ex) {
|
|
|
|
sink.error(ex);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-01-31 00:36:21 +01:00
|
|
|
@Override
|
2021-03-18 16:19:41 +01:00
|
|
|
public Mono<U> get(@Nullable CompositeSnapshot snapshot, boolean existsAlmostCertainly) {
|
2021-08-22 18:20:05 +02:00
|
|
|
return dictionary
|
|
|
|
.get(resolveSnapshot(snapshot), keyMono, existsAlmostCertainly)
|
2021-08-22 21:23:22 +02:00
|
|
|
.handle(this::deserializeValue);
|
2021-01-31 00:36:21 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
2021-01-31 19:52:47 +01:00
|
|
|
public Mono<U> setAndGetPrevious(U value) {
|
2021-08-29 23:18:03 +02:00
|
|
|
return dictionary
|
|
|
|
.put(keyMono, Mono.fromCallable(() -> serializer.serialize(value)), LLDictionaryResultType.PREVIOUS_VALUE)
|
|
|
|
.handle(this::deserializeValue);
|
2021-01-31 00:36:21 +01:00
|
|
|
}
|
|
|
|
|
2021-02-06 19:21:31 +01:00
|
|
|
@Override
|
2021-08-22 21:23:22 +02:00
|
|
|
public Mono<U> update(SerializationFunction<@Nullable U, @Nullable U> updater,
|
2021-05-08 03:09:00 +02:00
|
|
|
UpdateReturnMode updateReturnMode,
|
|
|
|
boolean existsAlmostCertainly) {
|
2021-08-22 18:20:05 +02:00
|
|
|
return dictionary
|
|
|
|
.update(keyMono, (oldValueSer) -> {
|
2021-09-08 00:22:39 +02:00
|
|
|
try (oldValueSer) {
|
|
|
|
var result = updater.apply(oldValueSer == null ? null
|
|
|
|
: serializer.deserialize(oldValueSer).deserializedData());
|
|
|
|
if (result == null) {
|
|
|
|
return null;
|
|
|
|
} else {
|
|
|
|
return serializer.serialize(result);
|
|
|
|
}
|
2021-05-12 19:02:51 +02:00
|
|
|
}
|
2021-08-22 18:20:05 +02:00
|
|
|
}, updateReturnMode, existsAlmostCertainly)
|
2021-08-22 21:23:22 +02:00
|
|
|
.handle(this::deserializeValue);
|
2021-05-08 03:09:00 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
2021-08-22 21:23:22 +02:00
|
|
|
public Mono<Delta<U>> updateAndGetDelta(SerializationFunction<@Nullable U, @Nullable U> updater,
|
2021-05-08 03:09:00 +02:00
|
|
|
boolean existsAlmostCertainly) {
|
2021-08-22 18:20:05 +02:00
|
|
|
return dictionary
|
|
|
|
.updateAndGetDelta(keyMono, (oldValueSer) -> {
|
2021-09-08 00:22:39 +02:00
|
|
|
try (oldValueSer) {
|
|
|
|
var result = updater.apply(oldValueSer == null ? null
|
|
|
|
: serializer.deserialize(oldValueSer).deserializedData());
|
|
|
|
if (result == null) {
|
|
|
|
return null;
|
|
|
|
} else {
|
|
|
|
return serializer.serialize(result);
|
|
|
|
}
|
2021-05-12 19:02:51 +02:00
|
|
|
}
|
2021-09-02 17:15:40 +02:00
|
|
|
}, existsAlmostCertainly).transform(mono -> LLUtils.mapLLDelta(mono,
|
|
|
|
serialized -> serializer.deserialize(serialized).deserializedData()
|
|
|
|
));
|
2021-02-06 19:21:31 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 00:36:21 +01:00
|
|
|
@Override
|
2021-01-31 19:52:47 +01:00
|
|
|
public Mono<U> clearAndGetPrevious() {
|
2021-08-22 18:20:05 +02:00
|
|
|
return dictionary
|
|
|
|
.remove(keyMono, LLDictionaryResultType.PREVIOUS_VALUE)
|
2021-08-22 21:23:22 +02:00
|
|
|
.handle(this::deserializeValue);
|
2021-01-31 00:36:21 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
2021-02-24 16:43:07 +01:00
|
|
|
public Mono<Long> leavesCount(@Nullable CompositeSnapshot snapshot, boolean fast) {
|
2021-08-22 18:20:05 +02:00
|
|
|
return dictionary
|
2021-08-29 23:18:03 +02:00
|
|
|
.isRangeEmpty(resolveSnapshot(snapshot), keyMono.map(LLRange::single).map(ResourceSupport::send))
|
2021-08-22 18:20:05 +02:00
|
|
|
.map(empty -> empty ? 0L : 1L);
|
2021-01-31 00:36:21 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public Mono<Boolean> isEmpty(@Nullable CompositeSnapshot snapshot) {
|
2021-08-22 18:20:05 +02:00
|
|
|
return dictionary
|
2021-08-29 23:18:03 +02:00
|
|
|
.isRangeEmpty(resolveSnapshot(snapshot), keyMono.map(LLRange::single).map(ResourceSupport::send));
|
2021-01-31 19:52:47 +01:00
|
|
|
}
|
2021-04-30 19:15:04 +02:00
|
|
|
|
|
|
|
@Override
|
|
|
|
public void release() {
|
2021-08-29 23:18:03 +02:00
|
|
|
key.close();
|
2021-04-30 19:15:04 +02:00
|
|
|
}
|
2021-06-26 02:35:33 +02:00
|
|
|
|
|
|
|
@Override
|
|
|
|
public Flux<BadBlock> badBlocks() {
|
2021-08-29 23:18:03 +02:00
|
|
|
return dictionary.badBlocks(keyMono.map(LLRange::single).map(ResourceSupport::send));
|
2021-06-26 02:35:33 +02:00
|
|
|
}
|
2021-01-31 00:36:21 +01:00
|
|
|
}
|