package it.cavallium.dbengine.database.collections; import io.net5.buffer.api.Buffer; 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; import it.cavallium.dbengine.client.BadBlock; import it.cavallium.dbengine.client.CompositeSnapshot; import it.cavallium.dbengine.database.Delta; import it.cavallium.dbengine.database.LLDictionary; import it.cavallium.dbengine.database.LLDictionaryResultType; import it.cavallium.dbengine.database.LLRange; import it.cavallium.dbengine.database.LLSnapshot; import it.cavallium.dbengine.database.LLUtils; import it.cavallium.dbengine.database.UpdateReturnMode; import it.cavallium.dbengine.database.serialization.SerializationException; import it.cavallium.dbengine.database.serialization.SerializationFunction; import it.cavallium.dbengine.database.serialization.Serializer; import org.jetbrains.annotations.Nullable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.SynchronousSink; public class DatabaseSingle extends ResourceSupport, DatabaseSingle> implements DatabaseStageEntry { private final LLDictionary dictionary; private final Mono> keyMono; private final Serializer serializer; private Buffer key; public DatabaseSingle(LLDictionary dictionary, Send key, Serializer serializer, Drop> drop) { super(new CloseOnDrop<>(drop)); try (key) { this.dictionary = dictionary; this.key = key.receive(); this.keyMono = LLUtils.lazyRetain(this.key); this.serializer = serializer; } } private LLSnapshot resolveSnapshot(@Nullable CompositeSnapshot snapshot) { if (snapshot == null) { return null; } else { return snapshot.getSnapshot(dictionary); } } private void deserializeValue(Send value, SynchronousSink sink) { try (value) { sink.next(serializer.deserialize(value).deserializedData()); } catch (SerializationException ex) { sink.error(ex); } } @Override public Mono get(@Nullable CompositeSnapshot snapshot, boolean existsAlmostCertainly) { return dictionary .get(resolveSnapshot(snapshot), keyMono, existsAlmostCertainly) .handle(this::deserializeValue); } @Override public Mono setAndGetPrevious(U value) { return dictionary .put(keyMono, Mono.fromCallable(() -> serializer.serialize(value)), LLDictionaryResultType.PREVIOUS_VALUE) .handle(this::deserializeValue); } @Override public Mono update(SerializationFunction<@Nullable U, @Nullable U> updater, UpdateReturnMode updateReturnMode, boolean existsAlmostCertainly) { return dictionary .update(keyMono, (oldValueSer) -> { try (oldValueSer) { var result = updater.apply(oldValueSer == null ? null : serializer.deserialize(oldValueSer).deserializedData()); if (result == null) { return null; } else { return serializer.serialize(result); } } }, updateReturnMode, existsAlmostCertainly) .handle(this::deserializeValue); } @Override public Mono> updateAndGetDelta(SerializationFunction<@Nullable U, @Nullable U> updater, boolean existsAlmostCertainly) { return dictionary .updateAndGetDelta(keyMono, (oldValueSer) -> { try (oldValueSer) { var result = updater.apply(oldValueSer == null ? null : serializer.deserialize(oldValueSer).deserializedData()); if (result == null) { return null; } else { return serializer.serialize(result); } } }, existsAlmostCertainly).transform(mono -> LLUtils.mapLLDelta(mono, serialized -> serializer.deserialize(serialized).deserializedData() )); } @Override public Mono clearAndGetPrevious() { return dictionary .remove(keyMono, LLDictionaryResultType.PREVIOUS_VALUE) .handle(this::deserializeValue); } @Override public Mono leavesCount(@Nullable CompositeSnapshot snapshot, boolean fast) { return dictionary .isRangeEmpty(resolveSnapshot(snapshot), keyMono.map(LLRange::single).map(ResourceSupport::send)) .map(empty -> empty ? 0L : 1L); } @Override public Mono isEmpty(@Nullable CompositeSnapshot snapshot) { return dictionary .isRangeEmpty(resolveSnapshot(snapshot), keyMono.map(LLRange::single).map(ResourceSupport::send)); } @Override public Flux badBlocks() { return dictionary.badBlocks(keyMono.map(LLRange::single).map(ResourceSupport::send)); } @Override protected RuntimeException createResourceClosedException() { throw new IllegalStateException("Closed"); } @Override protected Owned> prepareSend() { var key = this.key.send(); return drop -> new DatabaseSingle<>(dictionary, key, serializer, drop); } @Override protected void makeInaccessible() { this.key = null; } private static class CloseOnDrop implements Drop> { private final Drop> delegate; public CloseOnDrop(Drop> drop) { this.delegate = drop; } @Override public void drop(DatabaseSingle obj) { if (obj.key != null) { obj.key.close(); } delegate.drop(obj); } } }