2021-01-30 20:16:14 +01:00
|
|
|
package it.cavallium.dbengine.database.collections;
|
|
|
|
|
2021-01-31 12:07:07 +01:00
|
|
|
import io.netty.buffer.Unpooled;
|
2021-01-30 20:16:14 +01:00
|
|
|
import it.cavallium.dbengine.client.CompositeSnapshot;
|
|
|
|
import it.cavallium.dbengine.database.LLDictionary;
|
2021-01-31 21:23:43 +01:00
|
|
|
import it.cavallium.dbengine.database.LLDictionaryResultType;
|
|
|
|
import it.cavallium.dbengine.database.LLUtils;
|
2021-01-30 20:16:14 +01:00
|
|
|
import java.util.Arrays;
|
2021-01-31 21:23:43 +01:00
|
|
|
import java.util.HashMap;
|
2021-01-30 20:16:14 +01:00
|
|
|
import java.util.Map;
|
|
|
|
import java.util.Map.Entry;
|
|
|
|
import org.jetbrains.annotations.Nullable;
|
|
|
|
import reactor.core.publisher.Flux;
|
|
|
|
import reactor.core.publisher.Mono;
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
/**
|
|
|
|
* Optimized implementation of "DatabaseMapDictionary with SubStageGetterSingle"
|
|
|
|
*/
|
|
|
|
public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U, DatabaseStageEntry<U>> {
|
2021-01-30 20:16:14 +01:00
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
private final Serializer<U> valueSerializer;
|
2021-01-30 20:16:14 +01:00
|
|
|
|
2021-02-01 10:52:33 +01:00
|
|
|
protected DatabaseMapDictionary(LLDictionary dictionary, byte[] prefixKey, SerializerFixedBinaryLength<T> keySuffixSerializer, Serializer<U> valueSerializer) {
|
2021-01-31 21:23:43 +01:00
|
|
|
super(dictionary, new SubStageGetterSingle<>(valueSerializer), keySuffixSerializer, prefixKey, 0);
|
|
|
|
this.valueSerializer = valueSerializer;
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
public static <T, U> DatabaseMapDictionary<T, U> simple(LLDictionary dictionary,
|
2021-02-01 10:52:33 +01:00
|
|
|
SerializerFixedBinaryLength<T> keySerializer,
|
2021-01-31 21:23:43 +01:00
|
|
|
Serializer<U> valueSerializer) {
|
|
|
|
return new DatabaseMapDictionary<>(dictionary, EMPTY_BYTES, keySerializer, valueSerializer);
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
public static <T, U> DatabaseMapDictionary<T, U> tail(LLDictionary dictionary,
|
2021-02-01 10:52:33 +01:00
|
|
|
SerializerFixedBinaryLength<T> keySuffixSerializer,
|
2021-01-31 21:23:43 +01:00
|
|
|
Serializer<U> valueSerializer,
|
|
|
|
byte[] prefixKey) {
|
|
|
|
return new DatabaseMapDictionary<>(dictionary, prefixKey, keySuffixSerializer, valueSerializer);
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
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;
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<Map<T, U>> get(@Nullable CompositeSnapshot snapshot) {
|
|
|
|
return dictionary
|
|
|
|
.getRange(resolveSnapshot(snapshot), range)
|
|
|
|
.collectMap(entry -> deserializeSuffix(stripPrefix(entry.getKey())), entry -> deserialize(entry.getValue()), HashMap::new);
|
2021-01-31 19:52:47 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<Map<T, U>> setAndGetPrevious(Map<T, U> 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);
|
2021-01-31 19:52:47 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
private Entry<byte[], byte[]> stripPrefix(Entry<byte[], byte[]> entry) {
|
|
|
|
byte[] keySuffix = stripPrefix(entry.getKey());
|
|
|
|
return Map.entry(keySuffix, entry.getValue());
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<Map<T, U>> clearAndGetPrevious() {
|
|
|
|
return dictionary
|
|
|
|
.setRange(range, Flux.empty(), true)
|
|
|
|
.collectMap(entry -> deserializeSuffix(stripPrefix(entry.getKey())), entry -> deserialize(entry.getValue()), HashMap::new);
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<Long> size(@Nullable CompositeSnapshot snapshot, boolean fast) {
|
|
|
|
return dictionary.sizeRange(resolveSnapshot(snapshot), range, true);
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<DatabaseStageEntry<U>> at(@Nullable CompositeSnapshot snapshot, T keySuffix) {
|
|
|
|
return Mono.just(new DatabaseSingle<>(dictionary, toKey(serializeSuffix(keySuffix)), Serializer.noopBytes())).map(entry -> new DatabaseSingleMapped<>(entry, valueSerializer));
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<U> getValue(@Nullable CompositeSnapshot snapshot, T keySuffix) {
|
|
|
|
return dictionary.get(resolveSnapshot(snapshot), toKey(serializeSuffix(keySuffix))).map(this::deserialize);
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<Void> putValue(T keySuffix, U value) {
|
|
|
|
return dictionary.put(toKey(serializeSuffix(keySuffix)), serialize(value), LLDictionaryResultType.VOID).then();
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<U> putValueAndGetPrevious(T keySuffix, U value) {
|
|
|
|
return dictionary
|
|
|
|
.put(toKey(serializeSuffix(keySuffix)), serialize(value), LLDictionaryResultType.PREVIOUS_VALUE)
|
|
|
|
.map(this::deserialize);
|
2021-01-31 00:36:21 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<Boolean> putValueAndGetStatus(T keySuffix, U value) {
|
|
|
|
return dictionary
|
|
|
|
.put(toKey(serializeSuffix(keySuffix)), serialize(value), LLDictionaryResultType.VALUE_CHANGED)
|
|
|
|
.map(LLUtils::responseToBoolean);
|
2021-01-31 00:36:21 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<Void> remove(T keySuffix) {
|
|
|
|
return dictionary.remove(toKey(serializeSuffix(keySuffix)), LLDictionaryResultType.VOID).then();
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<U> removeAndGetPrevious(T keySuffix) {
|
|
|
|
return dictionary
|
|
|
|
.remove(toKey(serializeSuffix(keySuffix)), LLDictionaryResultType.PREVIOUS_VALUE)
|
|
|
|
.map(this::deserialize);
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 21:23:43 +01:00
|
|
|
@Override
|
|
|
|
public Mono<Boolean> removeAndGetStatus(T keySuffix) {
|
|
|
|
return dictionary
|
|
|
|
.remove(toKey(serializeSuffix(keySuffix)), LLDictionaryResultType.VALUE_CHANGED)
|
|
|
|
.map(LLUtils::responseToBoolean);
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
2021-01-31 21:23:43 +01:00
|
|
|
public Flux<Entry<T, U>> getMulti(@Nullable CompositeSnapshot snapshot, Flux<T> keys) {
|
|
|
|
return dictionary
|
|
|
|
.getMulti(resolveSnapshot(snapshot), keys.map(keySuffix -> toKey(serializeSuffix(keySuffix))))
|
|
|
|
.map(entry -> Map.entry(deserializeSuffix(stripPrefix(entry.getKey())), deserialize(entry.getValue())));
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
|
|
|
|
2021-02-01 02:21:53 +01:00
|
|
|
@Override
|
|
|
|
public Mono<Void> putMulti(Flux<Entry<T, U>> entries) {
|
|
|
|
return dictionary
|
|
|
|
.putMulti(entries
|
|
|
|
.map(entry -> Map.entry(toKey(serializeSuffix(entry.getKey())), serialize(entry.getValue()))), false)
|
|
|
|
.then();
|
|
|
|
}
|
|
|
|
|
2021-01-30 20:16:14 +01:00
|
|
|
@Override
|
2021-01-31 21:23:43 +01:00
|
|
|
public Flux<Entry<T, DatabaseStageEntry<U>>> getAllStages(@Nullable CompositeSnapshot snapshot) {
|
|
|
|
return dictionary
|
2021-01-31 00:36:21 +01:00
|
|
|
.getRangeKeys(resolveSnapshot(snapshot), range)
|
2021-01-31 21:23:43 +01:00
|
|
|
.map(keySuffix -> Map.entry(deserializeSuffix(stripPrefix(keySuffix)),
|
|
|
|
new DatabaseSingleMapped<>(new DatabaseSingle<>(dictionary, toKey(stripPrefix(keySuffix)), Serializer.noopBytes()),
|
|
|
|
valueSerializer
|
|
|
|
)
|
|
|
|
));
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|
2021-01-31 12:02:02 +01:00
|
|
|
|
2021-01-31 15:47:48 +01:00
|
|
|
@Override
|
|
|
|
public Flux<Entry<T, U>> setAllValuesAndGetPrevious(Flux<Entry<T, U>> entries) {
|
2021-01-31 21:23:43 +01:00
|
|
|
var serializedEntries = entries
|
|
|
|
.map(entry -> Map.entry(toKey(serializeSuffix(entry.getKey())), serialize(entry.getValue())));
|
|
|
|
return dictionary.setRange(range, serializedEntries, true)
|
|
|
|
.map(entry -> Map.entry(deserializeSuffix(stripPrefix(entry.getKey())), deserialize(entry.getValue())));
|
2021-01-31 15:47:48 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 12:07:07 +01:00
|
|
|
//todo: temporary wrapper. convert the whole class to buffers
|
2021-01-31 21:23:43 +01:00
|
|
|
private U deserialize(byte[] bytes) {
|
|
|
|
var serialized = Unpooled.wrappedBuffer(bytes);
|
|
|
|
return valueSerializer.deserialize(serialized);
|
2021-01-31 12:02:02 +01:00
|
|
|
}
|
|
|
|
|
2021-01-31 12:07:07 +01:00
|
|
|
//todo: temporary wrapper. convert the whole class to buffers
|
2021-01-31 21:23:43 +01:00
|
|
|
private byte[] serialize(U bytes) {
|
|
|
|
var output = Unpooled.buffer();
|
|
|
|
valueSerializer.serialize(bytes, output);
|
|
|
|
output.resetReaderIndex();
|
|
|
|
int length = output.readableBytes();
|
|
|
|
var outputBytes = new byte[length];
|
|
|
|
output.getBytes(0, outputBytes, 0, length);
|
2021-01-31 12:07:07 +01:00
|
|
|
return outputBytes;
|
2021-01-31 12:02:02 +01:00
|
|
|
}
|
2021-01-30 20:16:14 +01:00
|
|
|
}
|