Bugfixes

src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java

@@ -451,7 +451,10 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
 									luceneQuery = mltQuery;
 								}
 
-								return luceneSearch(doDistributedPre,
+								return luceneQuery;
+							})
+							.subscribeOn(luceneQueryScheduler)
+							.map(luceneQuery -> luceneSearch(doDistributedPre,
 									indexSearcher,
 									queryParams.getLimit(),
 									queryParams.getMinCompetitiveScore().getNullable(),
@@ -460,9 +463,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
 									luceneQuery,
 									QueryParser.toSort(queryParams.getSort()),
 									QueryParser.toScoreMode(queryParams.getScoreMode())
-								);
+							))
-							})
-							.subscribeOn(luceneQueryScheduler)
 							.materialize()
 							.flatMap(signal -> {
 								if (signal.isOnComplete() || signal.isOnError()) {
@@ -507,7 +508,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
 						})
 						.subscribeOn(luceneQueryScheduler)
 						.flatMap(tuple -> Mono
-								.fromCallable(() -> {
+								.fromSupplier(() -> {
 									Query luceneQuery = tuple.getT1();
 									Sort luceneSort = tuple.getT2().orElse(null);
 									ScoreMode luceneScoreMode = tuple.getT3();
@@ -522,7 +523,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
 											luceneSort,
 											luceneScoreMode
 									);
-								}).subscribeOn(luceneQueryScheduler)
+								})
 						)
 						.materialize()
 						.flatMap(signal -> {
@@ -545,7 +546,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
 			Query luceneQuery,
 			Sort luceneSort,
 			ScoreMode luceneScoreMode) {
-		var searchFlux = Flux.<LLSignal>create(sink -> {
+		return new LLSearchResult(Flux.just(Flux.defer(() -> Flux.<LLSignal>create(sink -> {
 			AtomicBoolean cancelled = new AtomicBoolean();
 			AtomicLong requests = new AtomicLong();
 			Semaphore requestsAvailable = new Semaphore(0);
@@ -553,23 +554,24 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
 				cancelled.set(true);
 				requestsAvailable.release();
 			});
+			sink.onCancel(() -> {
+				cancelled.set(true);
+				requestsAvailable.release();
+			});
 			sink.onRequest(delta -> {
 				requests.addAndGet(delta);
 				requestsAvailable.release();
 			});
 
+			luceneQueryScheduler.schedule(() -> {
 				try {
-				//noinspection BlockingMethodInNonBlockingContext
 					requestsAvailable.acquire();
-				requestsAvailable.release();
 					if (!cancelled.get()) {
 						if (doDistributedPre) {
-						//noinspection BlockingMethodInNonBlockingContext
 							allowOnlyQueryParsingCollectorStreamSearcher.search(indexSearcher, luceneQuery);
 							sink.next(new LLTotalHitsCount(0L));
 						} else {
 							int boundedLimit = Math.max(0, limit > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) limit);
-						//noinspection BlockingMethodInNonBlockingContext
 							streamSearcher.search(indexSearcher,
 									luceneQuery,
 									boundedLimit,
@@ -579,48 +581,44 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
 									keyFieldName,
 									keyScore -> {
 										try {
-										while (requests.get() <= 0 && !cancelled.get()) {
+											if (requests.get() <= 0 && !cancelled.get()) {
 												requestsAvailable.acquire();
 											}
-										if (!cancelled.get()) {
+											if (cancelled.get()) {
-											requests.decrementAndGet();
-											sink.next(fixKeyScore(keyScore, scoreDivisor));
-											return HandleResult.CONTINUE;
-										} else {
 												return HandleResult.HALT;
 											}
+											requests.updateAndGet(n -> n > 0 ? n - 1 : 0);
+											sink.next(fixKeyScore(keyScore, scoreDivisor));
+											return HandleResult.CONTINUE;
 										} catch (Exception ex) {
 											sink.error(ex);
 											cancelled.set(true);
+											requestsAvailable.release();
 											return HandleResult.HALT;
 										}
 									},
 									totalHitsCount -> {
 										try {
-										while (requests.get() <= 0 && !cancelled.get()) {
+											if (requests.get() <= 0 && !cancelled.get()) {
 												requestsAvailable.acquire();
 											}
-										if (!cancelled.get()) {
+											requests.updateAndGet(n -> n > 0 ? n - 1 : 0);
-											requests.decrementAndGet();
 											sink.next(new LLTotalHitsCount(totalHitsCount));
-										}
 										} catch (Exception ex) {
 											sink.error(ex);
 											cancelled.set(true);
+											requestsAvailable.release();
 										}
 									}
 							);
 						}
-					if (!cancelled.get()) {
 						sink.complete();
 					}
-				}
 				} catch (Exception ex) {
 					sink.error(ex);
 				}
-		}).subscribeOn(luceneQueryScheduler);
+			});
-
+		}).subscribeOn(Schedulers.boundedElastic()))));
-		return new LLSearchResult(Flux.just(searchFlux));
 	}
 
 	@Override

src/main/java/it/cavallium/dbengine/lucene/LuceneUtils.java

@@ -238,13 +238,14 @@ public class LuceneUtils {
 				.map(LuceneSignal::getTotalHitsCount)
 				.reduce(Long::sum)
 				.map(sum -> LuceneSignal.totalHitsCount(sum));
-		return sortedValues.mergeWith(sortedTotalSize);
+		return Flux.merge(sortedValues, sortedTotalSize);
 	}
 
 	public static Flux<LLSignal> mergeSignalStreamRaw(Flux<Flux<LLSignal>> mappedKeys,
 			MultiSort<LLSignal> mappedSort,
 			Long limit) {
 		Flux<Flux<LLSignal>> sharedMappedSignals = mappedKeys.publish().refCount(2);
+
 		Flux<LLSignal> sortedValues = LuceneUtils
 				.mergeStream(sharedMappedSignals.map(sub -> sub.filter(LLSignal::isValue)), mappedSort, limit);
 		//noinspection Convert2MethodRef
@@ -254,6 +255,7 @@ public class LuceneUtils {
 				.map(LLSignal::getTotalHitsCount)
 				.reduce(Long::sum)
 				.map(sum -> new LLTotalHitsCount(sum));
-		return sortedValues.mergeWith(sortedTotalSize);
+
+		return Flux.merge(sortedValues, sortedTotalSize);
 	}
 }