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

Add more discard handlers

Browse Source
This commit is contained in:
Andrea Cavalli 2021-11-08 12:06:32 +01:00
parent 59aa1ef5c6
commit 232e46bcea
6 changed files with 28 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/main/java/it/cavallium/dbengine/database/LLUtils.java
View File

@ -480,7 +480,8 @@ public class LLUtils {
return Mono.empty();
}
}, (r, ex) -> Mono.fromRunnable(() -> r.close()), r -> Mono.fromRunnable(() -> r.close()))
.doOnDiscard(Send.class, send -> send.close());
.doOnDiscard(Send.class, send -> send.close())
.doOnDiscard(Resource.class, Resource::close);
}
// todo: remove this ugly method
@ -916,6 +917,7 @@ public class LLUtils {
.doOnDiscard(Delta.class, LLUtils::discardDelta)
.doOnDiscard(LLDelta.class, LLUtils::discardLLDelta)
.doOnDiscard(Send.class, LLUtils::discardSend)
.doOnDiscard(Resource.class, LLUtils::discardResource)
.doOnDiscard(Map.class, LLUtils::discardMap)
.doOnDiscard(DatabaseStage.class, LLUtils::discardStage);
@ -944,6 +946,8 @@ public class LLUtils {
discardLLDelta(o);
} else if (obj instanceof Send o) {
discardSend(o);
} else if (obj instanceof Resource o) {
discardResource(o);
} else if (obj instanceof Map o) {
discardMap(o);
} else if (obj instanceof DatabaseStage o) {
@ -1059,6 +1063,10 @@ public class LLUtils {
send.close();
}
private static void discardResource(Resource<?> res) {
res.close();
}
private static void discardMap(Map<?, ?> map) {
for (Entry<?, ?> entry : map.entrySet()) {
boolean hasByteBuf = false;

5
src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionary.java
View File

@ -1,6 +1,7 @@
package it.cavallium.dbengine.database.collections;
import io.net5.buffer.api.Buffer;
import io.net5.buffer.api.Resource;
import io.net5.buffer.api.Send;
import io.net5.buffer.api.internal.ResourceSupport;
import it.cavallium.dbengine.client.CompositeSnapshot;
@ -364,11 +365,13 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
sink.error(e);
}
})
.doOnDiscard(Send.class, Send::close);
.doOnDiscard(Send.class, Send::close)
.doOnDiscard(Resource.class, Resource::close);
return dictionary
.putMulti(serializedEntries, false)
.then()
.doOnDiscard(Send.class, Send::close)
.doOnDiscard(Resource.class, Resource::close)
.doOnDiscard(LLEntry.class, ResourceSupport::close)
.doOnDiscard(List.class, list -> {
for (Object o : list) {

4
src/main/java/it/cavallium/dbengine/database/disk/CachedIndexSearcherManager.java
View File

@ -3,6 +3,7 @@ package it.cavallium.dbengine.database.disk;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import io.net5.buffer.api.Resource;
import io.net5.buffer.api.Send;
import it.cavallium.dbengine.database.LLSnapshot;
import java.io.IOException;
@ -148,7 +149,8 @@ public class CachedIndexSearcherManager implements IndexSearcherManager {
assert indexSearcher.getIndexReader().getRefCount() > 0;
return new LLIndexSearcher(indexSearcher, decRef, this::dropCachedIndexSearcher).send();
})
.doOnDiscard(Send.class, Send::close);
.doOnDiscard(Send.class, Send::close)
.doOnDiscard(Resource.class, Resource::close);
});
}

6
src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java
View File

@ -384,13 +384,15 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
return localSearcher
.collect(searcher, localQueryParams, keyFieldName, NO_TRANSFORMATION)
.map(result -> new LLSearchResultShard(result.results(), result.totalHitsCount(), result::close))
.doOnDiscard(Send.class, Send::close);
.doOnDiscard(Send.class, Send::close)
.doOnDiscard(Resource.class, Resource::close);
}
public Mono<Send<LLIndexSearcher>> retrieveSearcher(@Nullable LLSnapshot snapshot) {
return searcherManager
.retrieveSearcher(snapshot)
.doOnDiscard(Send.class, Send::close);
.doOnDiscard(Send.class, Send::close)
.doOnDiscard(Resource.class, Resource::close);
}
@Override

7
src/main/java/it/cavallium/dbengine/database/memory/LLMemoryDictionary.java
View File

@ -2,6 +2,7 @@ package it.cavallium.dbengine.database.memory;
import io.net5.buffer.api.Buffer;
import io.net5.buffer.api.BufferAllocator;
import io.net5.buffer.api.Resource;
import io.net5.buffer.api.Send;
import it.cavallium.dbengine.client.BadBlock;
import it.cavallium.dbengine.database.LLDelta;
@ -528,7 +529,8 @@ public class LLMemoryDictionary implements LLDictionary {
return getRange(snapshot, rangeMono)
.take(1, true)
.singleOrEmpty()
.doOnDiscard(Send.class, Send::close);
.doOnDiscard(Send.class, Send::close)
.doOnDiscard(Resource.class, Resource::close);
}
@Override
@ -536,7 +538,8 @@ public class LLMemoryDictionary implements LLDictionary {
return getRangeKeys(snapshot, rangeMono)
.take(1, true)
.singleOrEmpty()
.doOnDiscard(Send.class, Send::close);
.doOnDiscard(Send.class, Send::close)
.doOnDiscard(Resource.class, Resource::close);
}
@Override

4
src/main/java/it/cavallium/dbengine/lucene/searcher/CountLocalSearcher.java
View File

@ -1,5 +1,6 @@
package it.cavallium.dbengine.lucene.searcher;
import io.net5.buffer.api.Resource;
import io.net5.buffer.api.Send;
import it.cavallium.dbengine.client.query.current.data.TotalHitsCount;
import it.cavallium.dbengine.database.LLUtils;
@ -39,7 +40,8 @@ public class CountLocalSearcher implements LocalSearcher {
is -> Mono.empty()
)
.map(count -> new LuceneSearchResult(TotalHitsCount.of(count, true), Flux.empty(), null))
.doOnDiscard(Send.class, Send::close);
.doOnDiscard(Send.class, Send::close)
.doOnDiscard(Resource.class, Resource::close);
}
@Override