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

Fix some possible leaks

This commit is contained in:
Andrea Cavalli 2022-05-26 13:13:14 +02:00
parent 96de3023a0
commit 6056eedd75
4 changed files with 90 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/it/cavallium/dbengine/database/collections/DatabaseInt.java
View File

@ -20,13 +20,11 @@ public class DatabaseInt implements LLKeyValueDatabaseStructure {
}
public Mono<Integer> get(@Nullable LLSnapshot snapshot) {
return singleton.get(snapshot).handle((data, sink) -> {
try (data) {
sink.next(serializer.deserialize(data));
} catch (SerializationException e) {
sink.error(e);
}
});
var resultMono = singleton.get(snapshot);
return Mono.usingWhen(resultMono,
result -> Mono.fromSupplier(() -> serializer.deserialize(result)),
result -> Mono.fromRunnable(result::close)
);
}
public Mono<Void> set(int value) {

58
src/main/java/it/cavallium/dbengine/database/collections/DatabaseLong.java
View File

@ -24,17 +24,17 @@ public class DatabaseLong implements LLKeyValueDatabaseStructure {
}
public Mono<Long> get(@Nullable LLSnapshot snapshot) {
return singleton.get(snapshot).handle((data, sink) -> {
try (data) {
if (data.readableBytes() == 4) {
sink.next((long) (int) bugSerializer.deserialize(data));
} else {
sink.next(serializer.deserialize(data));
}
} catch (SerializationException e) {
sink.error(e);
}
});
var resultMono = singleton.get(snapshot);
return Mono.usingWhen(resultMono,
result -> Mono.fromSupplier(() -> {
if (result.readableBytes() == 4) {
return (long) (int) bugSerializer.deserialize(result);
} else {
return serializer.deserialize(result);
}
}),
result -> Mono.fromRunnable(result::close)
);
}
public Mono<Long> incrementAndGet() {
@ -62,24 +62,26 @@ public class DatabaseLong implements LLKeyValueDatabaseStructure {
}
private Mono<Long> addAnd(long count, UpdateReturnMode updateReturnMode) {
return singleton.update(prev -> {
if (prev != null) {
var prevLong = prev.readLong();
var alloc = singleton.getAllocator();
var buf = alloc.allocate(Long.BYTES);
buf.writeLong(prevLong + count);
return buf;
} else {
var alloc = singleton.getAllocator();
var buf = alloc.allocate(Long.BYTES);
buf.writeLong(count);
return buf;
var resultMono = singleton.update(prev -> {
try (prev) {
if (prev != null) {
var prevLong = prev.readLong();
var alloc = singleton.getAllocator();
var buf = alloc.allocate(Long.BYTES);
buf.writeLong(prevLong + count);
return buf;
} else {
var alloc = singleton.getAllocator();
var buf = alloc.allocate(Long.BYTES);
buf.writeLong(count);
return buf;
}
}
}, updateReturnMode).map(buf -> {
try (buf) {
return buf.readLong();
}
}).single();
}, updateReturnMode);
return Mono.usingWhen(resultMono,
result -> Mono.fromSupplier(result::readLong),
result -> Mono.fromRunnable(result::close)
).single();
}
public Mono<Void> set(long value) {

24
src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapSingle.java
View File

@ -130,17 +130,19 @@ public class DatabaseMapSingle<U> extends ResourceSupport<DatabaseStage<U>, Data
UpdateReturnMode updateReturnMode) {
var resultMono = dictionary
.update(keyMono, (oldValueSer) -> {
U result;
if (oldValueSer == null) {
result = updater.apply(null);
} else {
U deserializedValue = serializer.deserialize(oldValueSer);
result = updater.apply(deserializedValue);
}
if (result == null) {
return null;
} else {
return serializeValue(result);
try (oldValueSer) {
U result;
if (oldValueSer == null) {
result = updater.apply(null);
} else {
U deserializedValue = serializer.deserialize(oldValueSer);
result = updater.apply(deserializedValue);
}
if (result == null) {
return null;
} else {
return serializeValue(result);
}
}
}, updateReturnMode);
return Mono.usingWhen(resultMono,

76
src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleton.java
View File

@ -67,24 +67,22 @@ public class DatabaseSingleton<U> extends ResourceSupport<DatabaseStage<U>, Data
}
}
private void deserializeValue(Buffer value, SynchronousSink<U> sink) {
private U deserializeValue(Buffer value) {
try {
U deserializedValue;
try (value) {
deserializedValue = serializer.deserialize(value);
}
sink.next(deserializedValue);
return deserializedValue;
} catch (IndexOutOfBoundsException ex) {
var exMessage = ex.getMessage();
if (exMessage != null && exMessage.contains("read 0 to 0, write 0 to ")) {
LOG.error("Unexpected zero-bytes value at "
+ singleton.getDatabaseName() + ":" + singleton.getColumnName() + ":" + singleton.getName());
sink.complete();
LOG.error("Unexpected zero-bytes value at " + singleton.getDatabaseName()
+ ":" + singleton.getColumnName() + ":" + singleton.getName());
return null;
} else {
sink.error(ex);
throw ex;
}
} catch (SerializationException ex) {
sink.error(ex);
}
}
@ -103,8 +101,11 @@ public class DatabaseSingleton<U> extends ResourceSupport<DatabaseStage<U>, Data
@Override
public Mono<U> get(@Nullable CompositeSnapshot snapshot) {
return singleton.get(resolveSnapshot(snapshot))
.handle(this::deserializeValue);
var resultMono = singleton.get(resolveSnapshot(snapshot));
return Mono.usingWhen(resultMono,
result -> Mono.fromSupplier(() -> this.deserializeValue(result)),
result -> Mono.fromRunnable(result::close)
);
}
@Override
@ -114,33 +115,39 @@ public class DatabaseSingleton<U> extends ResourceSupport<DatabaseStage<U>, Data
@Override
public Mono<U> setAndGetPrevious(U value) {
return Flux
var resultMono = Flux
.concat(singleton.get(null), singleton.set(Mono.fromCallable(() -> serializeValue(value))).then(Mono.empty()))
.last()
.handle(this::deserializeValue);
.last();
return Mono.usingWhen(resultMono,
result -> Mono.fromSupplier(() -> this.deserializeValue(result)),
result -> Mono.fromRunnable(result::close)
);
}
@Override
public Mono<U> update(SerializationFunction<@Nullable U, @Nullable U> updater,
UpdateReturnMode updateReturnMode) {
return singleton
.update((oldValueSer) -> {
try (oldValueSer) {
U result;
if (oldValueSer == null) {
result = updater.apply(null);
} else {
U deserializedValue = serializer.deserialize(oldValueSer);
result = updater.apply(deserializedValue);
}
if (result == null) {
return null;
} else {
return serializeValue(result);
}
var resultMono = singleton
.update((oldValueSer) -> {
try (oldValueSer) {
U result;
if (oldValueSer == null) {
result = updater.apply(null);
} else {
U deserializedValue = serializer.deserialize(oldValueSer);
result = updater.apply(deserializedValue);
}
}, updateReturnMode)
.handle(this::deserializeValue);
if (result == null) {
return null;
} else {
return serializeValue(result);
}
}
}, updateReturnMode);
return Mono.usingWhen(resultMono,
result -> Mono.fromSupplier(() -> this.deserializeValue(result)),
result -> Mono.fromRunnable(result::close)
);
}
@Override
@ -171,10 +178,11 @@ public class DatabaseSingleton<U> extends ResourceSupport<DatabaseStage<U>, Data
@Override
public Mono<U> clearAndGetPrevious() {
return Flux
.concat(singleton.get(null), singleton.set(Mono.empty()).then(Mono.empty()))
.last()
.handle(this::deserializeValue);
var resultMono = Flux.concat(singleton.get(null), singleton.set(Mono.empty()).then(Mono.empty())).last();
return Mono.usingWhen(resultMono,
result -> Mono.fromSupplier(() -> this.deserializeValue(result)),
result -> Mono.fromRunnable(result::close)
);
}
@Override