andreacavalli/rockserver
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Execute all events in a dedicated executor

Browse Source
This commit is contained in:
Andrea Cavalli 2024-09-26 17:50:41 +02:00
parent 4d46a32bb4
commit 215ba242fd
2 changed files with 20 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/it/cavallium/rockserver/core/impl/rocksdb/SSTWriter.java
View File

@ -102,8 +102,8 @@ public record SSTWriter(RocksDB db, it.cavallium.rockserver.core.impl.ColumnInst
@Override
public void writePending() throws it.cavallium.rockserver.core.common.RocksDBException {
try {
checkOwningHandle();
try (this) {
checkOwningHandle();
sstFileWriter.finish();
try (var ingestOptions = new IngestExternalFileOptions()) {
ingestOptions

34
src/main/java/it/cavallium/rockserver/core/server/GrpcServer.java
View File

@ -290,25 +290,29 @@ public class GrpcServer extends Server {
@Override
public void onError(Throwable throwable) {
state = State.RECEIVED_ALL;
doFinally();
if (putBatchInputsSubscriber != null) {
putBatchInputsSubscriber.onError(throwable);
} else {
serverCallStreamObserver.onError(throwable);
}
sstExecutor.execute(() -> {
state = State.RECEIVED_ALL;
doFinally();
if (putBatchInputsSubscriber != null) {
putBatchInputsSubscriber.onError(throwable);
} else {
serverCallStreamObserver.onError(throwable);
}
});
}
@Override
public void onCompleted() {
if (state == State.BEFORE_INITIAL_REQUEST) {
serverCallStreamObserver.onError(RocksDBException.of(RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "Missing initial request"));
} else if (state == State.RECEIVING_DATA) {
state = State.RECEIVED_ALL;
checkCompleted(false);
} else {
putBatchInputsSubscriber.onError(RocksDBException.of(RocksDBException.RocksDBErrorType.PUT_UNKNOWN_ERROR, "Unknown state during onComplete: " + state));
}
sstExecutor.execute(() -> {
if (state == State.BEFORE_INITIAL_REQUEST) {
serverCallStreamObserver.onError(RocksDBException.of(RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "Missing initial request"));
} else if (state == State.RECEIVING_DATA) {
state = State.RECEIVED_ALL;
checkCompleted(false);
} else {
putBatchInputsSubscriber.onError(RocksDBException.of(RocksDBException.RocksDBErrorType.PUT_UNKNOWN_ERROR, "Unknown state during onComplete: " + state));
}
});
}
private void checkCompleted(boolean requestDone) {