andreacavalli/CavalliumDBEngine
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Deduplicate code

Browse Source
This commit is contained in:
Andrea Cavalli 2021-01-30 00:35:03 +01:00
parent 28614db942
commit 74fdb752b4
1 changed files with 19 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
src/main/java/it/cavallium/dbengine/database/disk/LLLocalDictionary.java
View File

@ -152,33 +152,7 @@ public class LLLocalDictionary implements LLDictionary {
@Override @Override
public Mono<byte[]> put(byte[] key, byte[] value, LLDictionaryResultType resultType) { public Mono<byte[]> put(byte[] key, byte[] value, LLDictionaryResultType resultType) {
Mono<byte[]> response = null; Mono<byte[]> response = getPrevValue(key, resultType);
switch (resultType) {
case VALUE_CHANGED:
response = containsKey(null, key).single().map(LLUtils::booleanToResponse);
break;
case PREVIOUS_VALUE:
response = Mono
.fromCallable(() -> {
var data = new Holder<byte[]>();
if (db.keyMayExist(cfh, key, data)) {
if (data.getValue() != null) {
return data.getValue();
} else {
return db.get(cfh, key);
}
} else {
return null;
}
})
.onErrorMap(IOException::new)
.subscribeOn(Schedulers.boundedElastic());
break;
case VOID:
response = Mono.empty();
break;
}
return Mono return Mono
.fromCallable(() -> { .fromCallable(() -> {
db.put(cfh, key, value); db.put(cfh, key, value);
@ -191,13 +165,23 @@ public class LLLocalDictionary implements LLDictionary {
@Override @Override
public Mono<byte[]> remove(byte[] key, LLDictionaryResultType resultType) { public Mono<byte[]> remove(byte[] key, LLDictionaryResultType resultType) {
Mono<byte[]> response = null; Mono<byte[]> response = getPrevValue(key, resultType);
return Mono
.fromCallable(() -> {
db.delete(cfh, key);
return null;
})
.onErrorMap(IOException::new)
.subscribeOn(Schedulers.boundedElastic())
.then(response);
}
private Mono<byte[]> getPrevValue(byte[] key, LLDictionaryResultType resultType) {
switch (resultType) { switch (resultType) {
case VALUE_CHANGED: case VALUE_CHANGED:
response = containsKey(null, key).single().map(LLUtils::booleanToResponse); return containsKey(null, key).single().map(LLUtils::booleanToResponse);
break;
case PREVIOUS_VALUE: case PREVIOUS_VALUE:
response = Mono return Mono
.fromCallable(() -> { .fromCallable(() -> {
var data = new Holder<byte[]>(); var data = new Holder<byte[]>();
if (db.keyMayExist(cfh, key, data)) { if (db.keyMayExist(cfh, key, data)) {
@ -212,20 +196,11 @@ public class LLLocalDictionary implements LLDictionary {
}) })
.onErrorMap(IOException::new) .onErrorMap(IOException::new)
.subscribeOn(Schedulers.boundedElastic()); .subscribeOn(Schedulers.boundedElastic());
break;
case VOID: case VOID:
response = Mono.empty(); return Mono.empty();
break; default:
return Mono.error(new IllegalStateException("Unexpected value: " + resultType));
} }
return Mono
.fromCallable(() -> {
db.delete(cfh, key);
return null;
})
.onErrorMap(IOException::new)
.subscribeOn(Schedulers.boundedElastic())
.then(response);
} }
@Override @Override
@ -268,7 +243,7 @@ public class LLLocalDictionary implements LLDictionary {
private Mono<Entry<byte[], byte[]>> putEntryToWriteBatch(Entry<byte[], byte[]> newEntry, boolean getOldValues, private Mono<Entry<byte[], byte[]>> putEntryToWriteBatch(Entry<byte[], byte[]> newEntry, boolean getOldValues,
CappedWriteBatch writeBatch) { CappedWriteBatch writeBatch) {
return Mono.from(Mono return Mono.from(Mono
.<byte[]>defer(() -> { .defer(() -> {
if (getOldValues) { if (getOldValues) {
return get(null, newEntry.getKey()); return get(null, newEntry.getKey());
} else { } else {