CavalliumDBEngine/src/test/java/it/cavallium/dbengine/TestDictionaryMap.java

744 lines
26 KiB
Java
Raw Normal View History

2021-05-02 19:18:15 +02:00
package it.cavallium.dbengine;
2021-05-02 19:18:15 +02:00
import static it.cavallium.dbengine.DbTestUtils.*;
import static it.cavallium.dbengine.SyncUtils.*;
2021-08-28 22:42:51 +02:00
import it.cavallium.dbengine.database.LLUtils;
import it.cavallium.dbengine.DbTestUtils.TestAllocator;
import it.cavallium.dbengine.database.UpdateMode;
import java.util.Arrays;
2021-05-08 03:09:00 +02:00
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
2021-08-28 22:42:51 +02:00
import org.junit.jupiter.api.AfterEach;
2021-09-01 00:01:56 +02:00
import org.junit.jupiter.api.Assertions;
2021-08-28 22:42:51 +02:00
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.warp.commonutils.log.Logger;
import org.warp.commonutils.log.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import reactor.test.StepVerifier.Step;
import reactor.test.util.TestLogger;
import reactor.util.Loggers;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
2021-09-06 01:15:13 +02:00
public abstract class TestDictionaryMap {
private static final Logger log = LoggerFactory.getLogger(TestDictionaryMap.class);
2021-08-29 01:15:51 +02:00
private TestAllocator allocator;
2021-09-08 00:22:39 +02:00
private boolean checkLeaks = true;
2021-08-29 01:15:51 +02:00
2021-05-02 19:18:15 +02:00
private static boolean isTestBadKeysEnabled() {
2021-09-10 13:34:26 +02:00
return !isCIMode() && System.getProperty("badkeys", "true").equalsIgnoreCase("true");
}
2021-09-06 01:15:13 +02:00
protected abstract TemporaryDbGenerator getTempDbGenerator();
private static Stream<Arguments> provideArgumentsPut() {
2021-05-08 03:09:00 +02:00
var goodKeys = List.of("12345");
List<String> badKeys;
2021-05-02 19:18:15 +02:00
if (isTestBadKeysEnabled()) {
2021-05-08 03:09:00 +02:00
badKeys = List.of("", "aaaa", "aaaaaa");
2021-05-02 19:18:15 +02:00
} else {
2021-05-08 03:09:00 +02:00
badKeys = List.of();
2021-05-02 19:18:15 +02:00
}
2021-05-08 03:09:00 +02:00
List<Tuple2<String, Boolean>> keys = Stream.concat(
goodKeys.stream().map(s -> Tuples.of(s, false)),
badKeys.stream().map(s -> Tuples.of(s, true))
2021-05-08 03:09:00 +02:00
).collect(Collectors.toList());
2021-09-10 13:34:26 +02:00
var values = isCIMode() ? List.of("val") : List.of("", "\0", BIG_STRING);
return keys
.stream()
.flatMap(keyTuple -> {
Stream<String> strm;
if (keyTuple.getT2()) {
2021-05-08 03:09:00 +02:00
strm = values.stream().findFirst().stream();
} else {
strm = values.stream();
}
return strm.map(val -> Tuples.of(keyTuple.getT1(), val, keyTuple.getT2()));
})
.flatMap(entryTuple -> Arrays.stream(UpdateMode.values()).map(updateMode -> Tuples.of(updateMode,
entryTuple.getT1(),
entryTuple.getT2(),
entryTuple.getT3()
)))
2021-09-06 01:15:13 +02:00
.flatMap(entryTuple -> Stream.of(Tuples.of(MapType.MAP, entryTuple.getT1(),
2021-05-03 18:07:18 +02:00
entryTuple.getT2(),
entryTuple.getT3(),
entryTuple.getT4()
2021-09-06 01:15:13 +02:00
), Tuples.of(MapType.HASH_MAP, entryTuple.getT1(),
2021-05-03 18:07:18 +02:00
entryTuple.getT2(),
entryTuple.getT3(),
2021-05-08 03:09:00 +02:00
false
2021-05-03 18:07:18 +02:00
)))
2021-09-06 01:15:13 +02:00
.filter(tuple -> !(tuple.getT1() == MapType.HASH_MAP && tuple.getT2() != UpdateMode.ALLOW))
2021-05-03 18:07:18 +02:00
.map(fullTuple -> Arguments.of(fullTuple.getT1(), fullTuple.getT2(), fullTuple.getT3(), fullTuple.getT4(), fullTuple.getT5()));
}
2021-08-28 22:42:51 +02:00
@BeforeEach
public void beforeEach() {
2021-08-29 01:15:51 +02:00
this.allocator = newAllocator();
ensureNoLeaks(allocator.allocator(), false, false);
2021-08-28 22:42:51 +02:00
}
@AfterEach
public void afterEach() {
2021-09-10 13:42:02 +02:00
if (!isCIMode() && checkLeaks) {
2021-09-08 00:22:39 +02:00
ensureNoLeaks(allocator.allocator(), true, false);
}
2021-08-29 01:15:51 +02:00
destroyAllocator(allocator);
2021-08-28 22:42:51 +02:00
}
@ParameterizedTest
@MethodSource("provideArgumentsPut")
2021-09-06 01:15:13 +02:00
public void testPut(MapType mapType, UpdateMode updateMode, String key, String value, boolean shouldFail) {
var gen = getTempDbGenerator();
var db = run(gen.openTempDb(allocator));
var dict = run(tempDictionary(db.db(), updateMode));
var map = tempDatabaseMapDictionaryMap(dict, mapType, 5);
runVoid(shouldFail, map.putValue(key, value));
var resultingMapSize = run(map.leavesCount(null, false));
Assertions.assertEquals(shouldFail ? 0 : 1, resultingMapSize);
var resultingMap = run(map.get(null));
Assertions.assertEquals(shouldFail ? null : Map.of(key, value), resultingMap);
2021-09-23 20:57:28 +02:00
map.close();
//if (shouldFail) this.checkLeaks = false;
gen.closeTempDb(db);
}
@ParameterizedTest
@MethodSource("provideArgumentsPut")
2021-09-06 01:15:13 +02:00
public void testAtSetAtGet(MapType mapType, UpdateMode updateMode, String key, String value, boolean shouldFail) {
var stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
2021-09-23 20:57:28 +02:00
.flatMap(map -> LLUtils
.usingResource(map.at(null, key), v -> v.set(value), true)
.then(LLUtils.usingResource(map.at(null, key), v -> v.get(null), true))
.doFinally(s -> map.close())
)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
stpVer.expectNext(value).verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPut")
2021-09-06 01:15:13 +02:00
public void testPutAndGetPrevious(MapType mapType, UpdateMode updateMode, String key, String value, boolean shouldFail) {
var stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.putValueAndGetPrevious(key, "error?"),
map.putValueAndGetPrevious(key, value),
map.putValueAndGetPrevious(key, value)
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
stpVer.expectNext("error?").expectNext(value).verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPut")
2021-09-06 01:15:13 +02:00
public void testPutValueRemoveAndGetPrevious(MapType mapType, UpdateMode updateMode, String key, String value, boolean shouldFail) {
var stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.removeAndGetPrevious(key),
map.putValue(key, value).then(map.removeAndGetPrevious(key)),
map.removeAndGetPrevious(key)
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
stpVer.expectNext(value).verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPut")
2021-09-06 01:15:13 +02:00
public void testPutValueRemoveAndGetStatus(MapType mapType, UpdateMode updateMode, String key, String value, boolean shouldFail) {
var stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.removeAndGetStatus(key),
map.putValue(key, value).then(map.removeAndGetStatus(key)),
map.removeAndGetStatus(key)
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
stpVer.expectNext(false, true, false).verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPut")
2021-09-06 01:15:13 +02:00
public void testUpdate(MapType mapType, UpdateMode updateMode, String key, String value, boolean shouldFail) {
2021-05-02 19:18:15 +02:00
if (updateMode == UpdateMode.DISALLOW && !isTestBadKeysEnabled()) {
return;
}
var stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.updateValue(key, old -> {
2021-10-19 00:22:05 +02:00
Assertions.assertNull(old);
return "error?";
}),
map.updateValue(key, false, old -> {
2021-09-01 00:01:56 +02:00
Assertions.assertEquals("error?", old);
return "error?";
}),
map.updateValue(key, true, old -> {
2021-09-01 00:01:56 +02:00
Assertions.assertEquals("error?", old);
return "error?";
}),
map.updateValue(key, true, old -> {
2021-09-01 00:01:56 +02:00
Assertions.assertEquals("error?", old);
return value;
}),
map.updateValue(key, true, old -> {
2021-09-01 00:01:56 +02:00
Assertions.assertEquals(value, old);
return value;
})
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-09-08 00:22:39 +02:00
.transform(LLUtils::handleDiscard)
));
if (updateMode == UpdateMode.DISALLOW || shouldFail) {
stpVer.verifyError();
} else {
stpVer.expectNext(true, false, false, true, false).verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPut")
2021-09-06 01:15:13 +02:00
public void testUpdateGet(MapType mapType, UpdateMode updateMode, String key, String value, boolean shouldFail) {
2021-05-02 19:18:15 +02:00
if (updateMode == UpdateMode.DISALLOW && !isTestBadKeysEnabled()) {
return;
}
var stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
Mono
.fromRunnable(() -> log.debug("1. Updating value: {}", key))
.then(map.updateValue(key, old -> {
assert old == null;
return "error?";
}))
.doOnSuccess(s -> log.debug("1. Getting value: {}", key))
.then(map.getValue(null, key)),
Mono
.fromRunnable(() -> log.debug("2. Updating value: {}", key))
.then(map.updateValue(key, false, old -> {
assert Objects.equals(old, "error?");
return "error?";
}))
.doOnSuccess(s -> log.debug("2. Getting value: {}", key))
.then(map.getValue(null, key)),
Mono
.fromRunnable(() -> log.debug("3. Updating value: {}", key))
.then(map.updateValue(key, true, old -> {
assert Objects.equals(old, "error?");
return "error?";
}))
.doOnSuccess(s -> log.debug("3. Getting value: {}", key))
.then(map.getValue(null, key)),
Mono
.fromRunnable(() -> log.debug("4. Updating value: {}", key))
.then(map.updateValue(key, true, old -> {
assert Objects.equals(old, "error?");
return value;
}))
.doOnSuccess(s -> log.debug("4. Getting value: {}", key))
.then(map.getValue(null, key)),
Mono
.fromRunnable(() -> log.debug("5. Updating value: {}", key))
.then(map.updateValue(key, true, old -> {
assert Objects.equals(old, value);
return value;
}))
.doOnSuccess(s -> log.debug("5. Getting value: {}", key))
.then(map.getValue(null, key))
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-09-08 00:22:39 +02:00
.transform(LLUtils::handleDiscard)
));
if (updateMode == UpdateMode.DISALLOW || shouldFail) {
stpVer.verifyError();
} else {
stpVer.expectNext("error?", "error?", "error?", value, value).verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPut")
2021-09-06 01:15:13 +02:00
public void testPutAndGetChanged(MapType mapType, UpdateMode updateMode, String key, String value, boolean shouldFail) {
var stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
2021-05-02 19:18:15 +02:00
map.putValueAndGetChanged(key, "error?").single(),
map.putValueAndGetChanged(key, value).single(),
map.putValueAndGetChanged(key, value).single(),
map.remove(key),
2021-05-02 19:18:15 +02:00
map.putValueAndGetChanged(key, "error?").single()
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
2021-05-02 19:18:15 +02:00
stpVer.expectNext(true, true, false, true).verifyComplete();
}
}
private static Stream<Arguments> provideArgumentsPutMulti() {
2021-09-10 13:34:26 +02:00
var goodKeys = isCIMode() ? List.of(List.of("12345")) : List.of(List.of("12345", "67890"), List.<String>of());
2021-05-08 03:09:00 +02:00
List<List<String>> badKeys;
2021-05-02 19:18:15 +02:00
if (isTestBadKeysEnabled()) {
2021-05-08 03:09:00 +02:00
badKeys = List.of(List.of("", "12345"), List.of("45678", "aaaa"), List.of("aaaaaa", "capra"));
2021-05-02 19:18:15 +02:00
} else {
2021-05-08 03:09:00 +02:00
badKeys = List.of();
2021-05-02 19:18:15 +02:00
}
2021-05-08 03:09:00 +02:00
List<Tuple2<List<String>, Boolean>> keys = Stream.concat(
goodKeys.stream().map(s -> Tuples.of(s, false)),
badKeys.stream().map(s -> Tuples.of(s, true))
2021-05-08 03:09:00 +02:00
).collect(Collectors.toList());
2021-09-10 13:34:26 +02:00
var values = isCIMode() ? List.of("val") : List.of("", "\0", BIG_STRING);
return keys
.stream()
.map(keyTuple -> keyTuple.mapT1(ks -> Flux
.zip(Flux.fromIterable(ks), Flux.fromIterable(values))
.collectMap(Tuple2::getT1, Tuple2::getT2)
.block()
))
.flatMap(entryTuple -> Arrays.stream(UpdateMode.values()).map(updateMode -> Tuples.of(updateMode,
entryTuple.getT1(),
entryTuple.getT2()
)))
2021-09-06 01:15:13 +02:00
.flatMap(entryTuple -> Stream.of(Tuples.of(MapType.MAP, entryTuple.getT1(),
2021-05-03 18:07:18 +02:00
entryTuple.getT2(),
entryTuple.getT3()
2021-09-06 01:15:13 +02:00
), Tuples.of(MapType.HASH_MAP, entryTuple.getT1(),
2021-05-03 18:07:18 +02:00
entryTuple.getT2(),
2021-05-08 03:09:00 +02:00
false
2021-05-03 18:07:18 +02:00
)))
2021-09-06 01:15:13 +02:00
.filter(tuple -> !(tuple.getT1() == MapType.HASH_MAP && tuple.getT2() != UpdateMode.ALLOW))
2021-05-03 18:07:18 +02:00
.map(fullTuple -> Arguments.of(fullTuple.getT1(), fullTuple.getT2(), fullTuple.getT3(), fullTuple.getT4()));
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testPutMultiGetMulti(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Entry<String, String>> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.putMulti(Flux.fromIterable(entries.entrySet())).then(Mono.empty()),
map.getMulti(null, Flux.fromIterable(entries.keySet()))
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-07-23 15:20:33 +02:00
.filter(k -> k.getValue().isPresent())
.map(k -> Map.entry(k.getKey(), k.getValue().orElseThrow()))
2021-08-28 22:42:51 +02:00
.transform(LLUtils::handleDiscard)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
entries.forEach((k, v) -> remainingEntries.add(Map.entry(k, v)));
for (Entry<String, String> ignored : remainingEntries) {
stpVer = stpVer.expectNextMatches(remainingEntries::remove);
}
stpVer.verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testSetAllValuesGetMulti(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Entry<String, String>> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> map
.setAllValues(Flux.fromIterable(entries.entrySet()))
.thenMany(map.getMulti(null, Flux.fromIterable(entries.keySet())))
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-07-23 15:20:33 +02:00
.filter(k -> k.getValue().isPresent())
.map(k -> Map.entry(k.getKey(), k.getValue().orElseThrow()))
2021-09-08 00:22:39 +02:00
.transform(LLUtils::handleDiscard)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
entries.forEach((k, v) -> remainingEntries.add(Map.entry(k, v)));
for (Entry<String, String> ignored : remainingEntries) {
stpVer = stpVer.expectNextMatches(remainingEntries::remove);
}
stpVer.verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testSetAllValuesAndGetPrevious(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Entry<String, String>> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.setAllValuesAndGetPrevious(Flux.fromIterable(entries.entrySet())),
map.setAllValuesAndGetPrevious(Flux.fromIterable(entries.entrySet()))
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-08-28 22:42:51 +02:00
.transform(LLUtils::handleDiscard)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
entries.forEach((k, v) -> remainingEntries.add(Map.entry(k, v)));
for (Entry<String, String> ignored : remainingEntries) {
stpVer = stpVer.expectNextMatches(remainingEntries::remove);
}
stpVer.verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testSetGetMulti(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Entry<String, String>> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.set(entries).then(Mono.empty()),
map.getMulti(null, Flux.fromIterable(entries.keySet()))
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-07-23 15:20:33 +02:00
.filter(k -> k.getValue().isPresent())
.map(k -> Map.entry(k.getKey(), k.getValue().orElseThrow()))
2021-09-08 00:22:39 +02:00
.transform(LLUtils::handleDiscard)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
entries.forEach((k, v) -> remainingEntries.add(Map.entry(k, v)));
for (Entry<String, String> ignored : remainingEntries) {
stpVer = stpVer.expectNextMatches(remainingEntries::remove);
}
stpVer.verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testSetAndGetChanged(MapType mapType,
2021-08-22 23:50:50 +02:00
UpdateMode updateMode,
Map<String, String> entries,
boolean shouldFail) {
Step<Boolean> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> {
Mono<Void> removalMono;
if (entries.isEmpty()) {
removalMono = Mono.empty();
} else {
2021-05-08 03:09:00 +02:00
removalMono = map.remove(entries.keySet().stream().findFirst().orElseThrow());
}
return Flux
.concat(
map.setAndGetChanged(entries).single(),
map.setAndGetChanged(entries).single(),
removalMono.then(Mono.empty()),
map.setAndGetChanged(entries).single()
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close());
})
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
stpVer.expectNext(!entries.isEmpty(), false, !entries.isEmpty()).verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testSetAndGetPrevious(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Entry<String, String>> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(map.setAndGetPrevious(entries), map.setAndGetPrevious(entries))
.map(Map::entrySet)
2021-07-31 18:00:53 +02:00
.concatMapIterable(list -> list)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
entries.forEach((k, v) -> remainingEntries.add(Map.entry(k, v)));
for (Entry<String, String> ignored : remainingEntries) {
stpVer = stpVer.expectNextMatches(remainingEntries::remove);
}
stpVer.verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testSetClearAndGetPreviousGet(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Entry<String, String>> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(map.set(entries).then(Mono.empty()), map.clearAndGetPrevious(), map.get(null))
.map(Map::entrySet)
2021-07-31 18:00:53 +02:00
.concatMapIterable(list -> list)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
entries.forEach((k, v) -> remainingEntries.add(Map.entry(k, v)));
for (Entry<String, String> ignored : remainingEntries) {
stpVer = stpVer.expectNextMatches(remainingEntries::remove);
}
stpVer.verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testPutMultiGetAllValues(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Entry<String, String>> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.putMulti(Flux.fromIterable(entries.entrySet())).then(Mono.empty()),
map.getAllValues(null)
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-08-28 22:42:51 +02:00
.transform(LLUtils::handleDiscard)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
entries.forEach((k, v) -> remainingEntries.add(Map.entry(k, v)));
for (Entry<String, String> ignored : remainingEntries) {
stpVer = stpVer.expectNextMatches(remainingEntries::remove);
}
stpVer.verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testPutMultiGet(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Entry<String, String>> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.putMulti(Flux.fromIterable(entries.entrySet())).then(Mono.empty()),
map.get(null)
2021-10-19 00:22:05 +02:00
.map(Map::entrySet)
.flatMapIterable(list -> list)
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-08-28 22:42:51 +02:00
.transform(LLUtils::handleDiscard)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
entries.forEach((k, v) -> remainingEntries.add(Map.entry(k, v)));
for (Entry<String, String> ignored : remainingEntries) {
stpVer = stpVer.expectNextMatches(remainingEntries::remove);
}
stpVer.verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testPutMultiGetAllStagesGet(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Entry<String, String>> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.putMulti(Flux.fromIterable(entries.entrySet())).then(Mono.empty()),
map
.getAllStages(null)
.flatMap(stage -> stage
.getValue()
.get(null)
.map(val -> Map.entry(stage.getKey(), val))
2021-09-23 20:57:28 +02:00
.doFinally(s -> stage.getValue().close())
)
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-08-28 22:42:51 +02:00
.transform(LLUtils::handleDiscard)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
stpVer.verifyError();
} else {
entries.forEach((k, v) -> remainingEntries.add(Map.entry(k, v)));
for (Entry<String, String> ignored : remainingEntries) {
stpVer = stpVer.expectNextMatches(remainingEntries::remove);
}
stpVer.verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testPutMultiIsEmpty(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true);
Step<Boolean> stpVer = StepVerifier
2021-09-06 01:15:13 +02:00
.create(tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.isEmpty(null),
map.putMulti(Flux.fromIterable(entries.entrySet())).then(Mono.empty()),
map.isEmpty(null)
)
2021-09-23 20:57:28 +02:00
.doFinally(s -> map.close())
)
2021-05-08 03:09:00 +02:00
.flatMap(val -> shouldFail ? Mono.empty() : Mono.just(val))
2021-08-28 22:42:51 +02:00
.transform(LLUtils::handleDiscard)
));
if (shouldFail) {
2021-09-08 00:22:39 +02:00
this.checkLeaks = false;
2021-05-08 03:09:00 +02:00
stpVer.verifyError();
} else {
stpVer.expectNext(true, entries.isEmpty()).verifyComplete();
}
}
@ParameterizedTest
@MethodSource("provideArgumentsPutMulti")
2021-09-06 01:15:13 +02:00
public void testPutMultiClear(MapType mapType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) {
2021-10-19 00:22:05 +02:00
List<Boolean> result;
try {
result = SyncUtils.run(DbTestUtils.tempDb(getTempDbGenerator(), allocator, db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, mapType, 5))
.flatMapMany(map -> Flux
.concat(
map.isEmpty(null),
map.putMulti(Flux.fromIterable(entries.entrySet())).then(Mono.empty()),
map.isEmpty(null),
map.clear().then(Mono.empty()),
map.isEmpty(null)
)
.doFinally(s -> map.close())
)
.flatMap(val -> shouldFail ? Mono.empty() : Mono.just(val))
.transform(LLUtils::handleDiscard)
.collectList()
).singleOrEmpty());
} catch (Exception ex) {
if (shouldFail) {
this.checkLeaks = false;
} else {
throw ex;
}
return;
}
2021-10-19 00:22:05 +02:00
Assertions.assertEquals(true, result.get(0));
Assertions.assertEquals(entries.isEmpty(), result.get(1));
Assertions.assertEquals(true, result.get(2));
}
}