package it.cavallium.dbengine.database.collections; 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.client.Mapper; import it.cavallium.dbengine.database.Delta; import it.cavallium.dbengine.database.LLEntry; 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 org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.jetbrains.annotations.Nullable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.SynchronousSink; @SuppressWarnings("unused") public class DatabaseSingleMapped extends ResourceSupport, DatabaseSingleMapped> implements DatabaseStageEntry { private static final Logger logger = LogManager.getLogger(DatabaseSingleMapped.class); private static final Drop> 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> fork() { return this; } @Override public void attach(DatabaseSingleMapped obj) { } }; private final Mapper mapper; private DatabaseStageEntry serializedSingle; @SuppressWarnings({"unchecked", "rawtypes"}) public DatabaseSingleMapped(DatabaseStageEntry serializedSingle, Mapper mapper, Drop> drop) { super((Drop>) (Drop) DROP); this.serializedSingle = serializedSingle; this.mapper = mapper; } @SuppressWarnings({"unchecked", "rawtypes"}) private DatabaseSingleMapped(Send> serializedSingle, Mapper mapper, Drop> drop) { super((Drop>) (Drop) DROP); this.mapper = mapper; this.serializedSingle = (DatabaseStageEntry) serializedSingle.receive(); } private void deserializeSink(B value, SynchronousSink sink) { try { sink.next(this.unMap(value)); } catch (SerializationException ex) { sink.error(ex); } } @Override public Mono get(@Nullable CompositeSnapshot snapshot, boolean existsAlmostCertainly) { return serializedSingle.get(snapshot, existsAlmostCertainly).handle(this::deserializeSink); } @Override public Mono getOrDefault(@Nullable CompositeSnapshot snapshot, Mono defaultValue) { return serializedSingle.get(snapshot).handle(this::deserializeSink).switchIfEmpty(defaultValue); } @Override public Mono set(A value) { return Mono .fromCallable(() -> map(value)) .flatMap(serializedSingle::set); } @Override public Mono setAndGetPrevious(A value) { return Mono .fromCallable(() -> map(value)) .flatMap(serializedSingle::setAndGetPrevious) .handle(this::deserializeSink); } @Override public Mono setAndGetChanged(A value) { return Mono .fromCallable(() -> map(value)) .flatMap(serializedSingle::setAndGetChanged) .single(); } @Override public Mono update(SerializationFunction<@Nullable A, @Nullable A> updater, UpdateReturnMode updateReturnMode, boolean existsAlmostCertainly) { return serializedSingle.update(oldValue -> { var result = updater.apply(oldValue == null ? null : this.unMap(oldValue)); if (result == null) { return null; } else { return this.map(result); } }, updateReturnMode, existsAlmostCertainly).handle(this::deserializeSink); } @Override public Mono> updateAndGetDelta(SerializationFunction<@Nullable A, @Nullable A> updater, boolean existsAlmostCertainly) { return serializedSingle.updateAndGetDelta(oldValue -> { var result = updater.apply(oldValue == null ? null : this.unMap(oldValue)); if (result == null) { return null; } else { return this.map(result); } }, existsAlmostCertainly).transform(mono -> LLUtils.mapDelta(mono, this::unMap)); } @Override public Mono clear() { return serializedSingle.clear(); } @Override public Mono clearAndGetPrevious() { return serializedSingle.clearAndGetPrevious().handle(this::deserializeSink); } @Override public Mono clearAndGetStatus() { return serializedSingle.clearAndGetStatus(); } @Override public Mono leavesCount(@Nullable CompositeSnapshot snapshot, boolean fast) { return serializedSingle.leavesCount(snapshot, fast); } @Override public Mono isEmpty(@Nullable CompositeSnapshot snapshot) { return serializedSingle.isEmpty(snapshot); } @Override public DatabaseStageEntry entry() { return this; } @Override public Flux badBlocks() { return this.serializedSingle.badBlocks(); } private A unMap(B bytes) throws SerializationException { return mapper.unmap(bytes); } private B map(A bytes) throws SerializationException { return mapper.map(bytes); } @Override protected RuntimeException createResourceClosedException() { throw new IllegalStateException("Closed"); } @Override protected Owned> prepareSend() { var serializedSingle = this.serializedSingle.send(); return drop -> new DatabaseSingleMapped<>(serializedSingle, mapper, drop); } @Override protected void makeInaccessible() { this.serializedSingle = null; } }