Bugfixes

src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionaryHashed.java

@@ -32,7 +32,7 @@ public class DatabaseSetDictionaryHashed<T, TH> extends DatabaseMapDictionaryHas
 	}
 
 	public static <T, TH> DatabaseSetDictionaryHashed<T, TH> simple(LLDictionary dictionary,
-			SerializerFixedBinaryLength<T, byte[]> keySerializer,
+			Serializer<T, byte[]> keySerializer,
 			Function<T, TH> keyHashFunction,
 			SerializerFixedBinaryLength<TH, byte[]> keyHashSerializer) {
 		return new DatabaseSetDictionaryHashed<>(dictionary,
@@ -45,7 +45,7 @@ public class DatabaseSetDictionaryHashed<T, TH> extends DatabaseMapDictionaryHas
 
 	public static <T, TH> DatabaseSetDictionaryHashed<T, TH> tail(LLDictionary dictionary,
 			byte[] prefixKey,
-			SerializerFixedBinaryLength<T, byte[]> keySuffixSerializer,
+			Serializer<T, byte[]> keySuffixSerializer,
 			Function<T, TH> keyHashFunction,
 			SerializerFixedBinaryLength<TH, byte[]> keyHashSerializer) {
 		return new DatabaseSetDictionaryHashed<>(dictionary,

src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashMap.java

@@ -0,0 +1,79 @@
+package it.cavallium.dbengine.database.collections;
+
+import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.LLDictionary;
+import it.cavallium.dbengine.database.serialization.Serializer;
+import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength;
+import java.util.Map;
+import java.util.function.Function;
+import org.jetbrains.annotations.Nullable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@SuppressWarnings("unused")
+public class SubStageGetterHashMap<T, U, TH> implements
+		SubStageGetter<Map<T, U>, DatabaseMapDictionaryHashed<T, U, TH>> {
+
+	private static final boolean assertsEnabled;
+	static {
+		boolean assertsEnabledTmp = false;
+		//noinspection AssertWithSideEffects
+		assert assertsEnabledTmp = true;
+		//noinspection ConstantConditions
+		assertsEnabled = assertsEnabledTmp;
+	}
+
+	private final Serializer<T, byte[]> keySerializer;
+	private final Serializer<U, byte[]> valueSerializer;
+	private final Function<T, TH> keyHashFunction;
+	private final SerializerFixedBinaryLength<TH, byte[]> keyHashSerializer;
+
+	public SubStageGetterHashMap(Serializer<T, byte[]> keySerializer,
+			Serializer<U, byte[]> valueSerializer,
+			Function<T, TH> keyHashFunction,
+			SerializerFixedBinaryLength<TH, byte[]> keyHashSerializer) {
+		this.keySerializer = keySerializer;
+		this.valueSerializer = valueSerializer;
+		this.keyHashFunction = keyHashFunction;
+		this.keyHashSerializer = keyHashSerializer;
+	}
+
+	@Override
+	public Mono<DatabaseMapDictionaryHashed<T, U, TH>> subStage(LLDictionary dictionary,
+			@Nullable CompositeSnapshot snapshot,
+			byte[] prefixKey,
+			Flux<byte[]> debuggingKeyFlux) {
+		Mono<DatabaseMapDictionaryHashed<T, U, TH>> result = Mono.just(DatabaseMapDictionaryHashed.tail(dictionary,
+				prefixKey,
+				keySerializer,
+				valueSerializer,
+				keyHashFunction,
+				keyHashSerializer
+		));
+		if (assertsEnabled) {
+			return checkKeyFluxConsistency(prefixKey, debuggingKeyFlux).then(result);
+		} else {
+			return result;
+		}
+	}
+
+	@Override
+	public boolean isMultiKey() {
+		return true;
+	}
+
+	@Override
+	public boolean needsDebuggingKeyFlux() {
+		return assertsEnabled;
+	}
+
+	private Mono<Void> checkKeyFluxConsistency(byte[] prefixKey, Flux<byte[]> keyFlux) {
+		return keyFlux.doOnNext(key -> {
+			assert key.length == prefixKey.length + getKeyHashBinaryLength();
+		}).then();
+	}
+
+	public int getKeyHashBinaryLength() {
+		return keyHashSerializer.getSerializedBinaryLength();
+	}
+}

src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashSet.java

@@ -0,0 +1,76 @@
+package it.cavallium.dbengine.database.collections;
+
+import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.LLDictionary;
+import it.cavallium.dbengine.database.collections.DatabaseEmpty.Nothing;
+import it.cavallium.dbengine.database.serialization.Serializer;
+import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength;
+import java.util.Map;
+import java.util.function.Function;
+import org.jetbrains.annotations.Nullable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@SuppressWarnings("unused")
+public class SubStageGetterHashSet<T, TH> implements
+		SubStageGetter<Map<T, Nothing>, DatabaseSetDictionaryHashed<T, TH>> {
+
+	private static final boolean assertsEnabled;
+	static {
+		boolean assertsEnabledTmp = false;
+		//noinspection AssertWithSideEffects
+		assert assertsEnabledTmp = true;
+		//noinspection ConstantConditions
+		assertsEnabled = assertsEnabledTmp;
+	}
+
+	private final Serializer<T, byte[]> keySerializer;
+	private final Function<T, TH> keyHashFunction;
+	private final SerializerFixedBinaryLength<TH, byte[]> keyHashSerializer;
+
+	public SubStageGetterHashSet(Serializer<T, byte[]> keySerializer,
+			Function<T, TH> keyHashFunction,
+			SerializerFixedBinaryLength<TH, byte[]> keyHashSerializer) {
+		this.keySerializer = keySerializer;
+		this.keyHashFunction = keyHashFunction;
+		this.keyHashSerializer = keyHashSerializer;
+	}
+
+	@Override
+	public Mono<DatabaseSetDictionaryHashed<T, TH>> subStage(LLDictionary dictionary,
+			@Nullable CompositeSnapshot snapshot,
+			byte[] prefixKey,
+			Flux<byte[]> debuggingKeyFlux) {
+		Mono<DatabaseSetDictionaryHashed<T, TH>> result = Mono.just(DatabaseSetDictionaryHashed.tail(dictionary,
+				prefixKey,
+				keySerializer,
+				keyHashFunction,
+				keyHashSerializer
+		));
+		if (assertsEnabled) {
+			return checkKeyFluxConsistency(prefixKey, debuggingKeyFlux).then(result);
+		} else {
+			return result;
+		}
+	}
+
+	@Override
+	public boolean isMultiKey() {
+		return true;
+	}
+
+	@Override
+	public boolean needsDebuggingKeyFlux() {
+		return assertsEnabled;
+	}
+
+	private Mono<Void> checkKeyFluxConsistency(byte[] prefixKey, Flux<byte[]> keyFlux) {
+		return keyFlux.doOnNext(key -> {
+			assert key.length == prefixKey.length + getKeyHashBinaryLength();
+		}).then();
+	}
+
+	public int getKeyHashBinaryLength() {
+		return keyHashSerializer.getSerializedBinaryLength();
+	}
+}