From 2fb96eaf039067e8281a224bf27ecd41ff319340 Mon Sep 17 00:00:00 2001 From: Andrea Cavalli Date: Wed, 29 Sep 2021 11:38:29 +0200 Subject: [PATCH] Ensure running in a blocking thread --- .../dbengine/lucene/collector/ReactiveLeafCollector.java | 2 ++ .../searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java | 1 + 2 files changed, 3 insertions(+) diff --git a/src/main/java/it/cavallium/dbengine/lucene/collector/ReactiveLeafCollector.java b/src/main/java/it/cavallium/dbengine/lucene/collector/ReactiveLeafCollector.java index 6eea953..6a04161 100644 --- a/src/main/java/it/cavallium/dbengine/lucene/collector/ReactiveLeafCollector.java +++ b/src/main/java/it/cavallium/dbengine/lucene/collector/ReactiveLeafCollector.java @@ -1,5 +1,6 @@ package it.cavallium.dbengine.lucene.collector; +import it.cavallium.dbengine.database.LLUtils; import java.util.concurrent.locks.LockSupport; import org.apache.lucene.index.LeafReaderContext; import org.apache.lucene.search.LeafCollector; @@ -27,6 +28,7 @@ public class ReactiveLeafCollector implements LeafCollector { @Override public void collect(int i) { + LLUtils.ensureBlocking(); var scoreDoc = new ScoreDoc(leafReaderContext.docBase + i, 0, shardIndex); boolean shouldRetry; do { diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java index 5881c7e..fe996df 100644 --- a/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java +++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java @@ -53,6 +53,7 @@ public class UnsortedUnscoredContinuousLuceneMultiSearcher implements LuceneMult return LLUtils.usingSendResource(indexSearchersSendResource, indexSearchers -> Mono.fromCallable(() -> { + LLUtils.ensureBlocking(); Many scoreDocsSink = Sinks.many().unicast().onBackpressureBuffer(QUEUE_SUPPLIER.get());