package it.cavallium.dbengine.database.collections; import it.cavallium.dbengine.client.CompositeSnapshot; import it.cavallium.dbengine.database.LLDictionary; import it.cavallium.dbengine.database.LLDictionaryResultType; import it.cavallium.dbengine.database.LLUtils; import it.cavallium.dbengine.database.serialization.Serializer; import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength; import java.util.Arrays; import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; import org.jetbrains.annotations.Nullable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; /** * Optimized implementation of "DatabaseMapDictionary with SubStageGetterSingle" */ public class DatabaseMapDictionary extends DatabaseMapDictionaryDeep> { private final Serializer valueSerializer; protected DatabaseMapDictionary(LLDictionary dictionary, byte[] prefixKey, SerializerFixedBinaryLength keySuffixSerializer, Serializer valueSerializer) { super(dictionary, prefixKey, keySuffixSerializer, new SubStageGetterSingle<>(valueSerializer), 0); this.valueSerializer = valueSerializer; } public static DatabaseMapDictionary simple(LLDictionary dictionary, SerializerFixedBinaryLength keySerializer, Serializer valueSerializer) { return new DatabaseMapDictionary<>(dictionary, EMPTY_BYTES, keySerializer, valueSerializer); } public static DatabaseMapDictionary tail(LLDictionary dictionary, byte[] prefixKey, SerializerFixedBinaryLength keySuffixSerializer, Serializer valueSerializer) { return new DatabaseMapDictionary<>(dictionary, prefixKey, keySuffixSerializer, valueSerializer); } private byte[] toKey(byte[] suffixKey) { assert suffixKeyConsistency(suffixKey.length); byte[] key = Arrays.copyOf(keyPrefix, keyPrefix.length + suffixKey.length); System.arraycopy(suffixKey, 0, key, keyPrefix.length, suffixKey.length); return key; } @Override public Mono> get(@Nullable CompositeSnapshot snapshot) { return dictionary .getRange(resolveSnapshot(snapshot), range) .collectMap( entry -> deserializeSuffix(stripPrefix(entry.getKey())), entry -> deserialize(entry.getValue()), HashMap::new); } @Override public Mono> setAndGetPrevious(Map value) { return dictionary .setRange(range, Flux .fromIterable(value.entrySet()) .map(entry -> Map.entry(serializeSuffix(entry.getKey()), serialize(entry.getValue()))), true ) .collectMap( entry -> deserializeSuffix(stripPrefix(entry.getKey())), entry -> deserialize(entry.getValue()), HashMap::new); } private Entry stripPrefix(Entry entry) { return Map.entry(stripPrefix(entry.getKey()), entry.getValue()); } @Override public Mono> clearAndGetPrevious() { return dictionary .setRange(range, Flux.empty(), true) .collectMap( entry -> deserializeSuffix(stripPrefix(entry.getKey())), entry -> deserialize(entry.getValue()), HashMap::new); } @Override public Mono size(@Nullable CompositeSnapshot snapshot, boolean fast) { return dictionary.sizeRange(resolveSnapshot(snapshot), range, fast); } @Override public Mono> at(@Nullable CompositeSnapshot snapshot, T keySuffix) { return Mono .just(new DatabaseSingle<>(dictionary, toKey(serializeSuffix(keySuffix)), Serializer.noop())) .map(entry -> new DatabaseSingleMapped<>(entry, valueSerializer)); } @Override public Mono getValue(@Nullable CompositeSnapshot snapshot, T keySuffix) { return dictionary.get(resolveSnapshot(snapshot), toKey(serializeSuffix(keySuffix))).map(this::deserialize); } @Override public Mono putValue(T keySuffix, U value) { return dictionary.put(toKey(serializeSuffix(keySuffix)), serialize(value), LLDictionaryResultType.VOID).then(); } @Override public Mono putValueAndGetPrevious(T keySuffix, U value) { return dictionary .put(toKey(serializeSuffix(keySuffix)), serialize(value), LLDictionaryResultType.PREVIOUS_VALUE) .map(this::deserialize); } @Override public Mono putValueAndGetStatus(T keySuffix, U value) { return dictionary .put(toKey(serializeSuffix(keySuffix)), serialize(value), LLDictionaryResultType.VALUE_CHANGED) .map(LLUtils::responseToBoolean); } @Override public Mono remove(T keySuffix) { return dictionary.remove(toKey(serializeSuffix(keySuffix)), LLDictionaryResultType.VOID).then(); } @Override public Mono removeAndGetPrevious(T keySuffix) { return dictionary .remove(toKey(serializeSuffix(keySuffix)), LLDictionaryResultType.PREVIOUS_VALUE) .map(this::deserialize); } @Override public Mono removeAndGetStatus(T keySuffix) { return dictionary .remove(toKey(serializeSuffix(keySuffix)), LLDictionaryResultType.VALUE_CHANGED) .map(LLUtils::responseToBoolean); } @Override public Flux> getMulti(@Nullable CompositeSnapshot snapshot, Flux keys) { return dictionary .getMulti(resolveSnapshot(snapshot), keys.map(keySuffix -> toKey(serializeSuffix(keySuffix)))) .map(entry -> Map.entry(deserializeSuffix(stripPrefix(entry.getKey())), deserialize(entry.getValue()))); } @Override public Mono putMulti(Flux> entries) { return dictionary .putMulti(entries .map(entry -> Map .entry(toKey(serializeSuffix(entry.getKey())), serialize(entry.getValue()))), false) .then(); } @Override public Flux>> getAllStages(@Nullable CompositeSnapshot snapshot) { return dictionary .getRangeKeys(resolveSnapshot(snapshot), range) .map(keySuffix -> Map.entry(deserializeSuffix(stripPrefix(keySuffix)), new DatabaseSingleMapped<>( new DatabaseSingle<>(dictionary, toKey(stripPrefix(keySuffix)), Serializer.noop()), valueSerializer ) )); } @Override public Flux> setAllValuesAndGetPrevious(Flux> entries) { return dictionary .setRange(range, entries.map(entry -> Map.entry(toKey(serializeSuffix(entry.getKey())), serialize(entry.getValue()))), true) .map(entry -> Map.entry(deserializeSuffix(stripPrefix(entry.getKey())), deserialize(entry.getValue()))); } //todo: temporary wrapper. convert the whole class to buffers private U deserialize(byte[] bytes) { return valueSerializer.deserialize(bytes); } //todo: temporary wrapper. convert the whole class to buffers private byte[] serialize(U bytes) { return valueSerializer.serialize(bytes); } }