161sh/CatgirlIntelligenceAgency
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

(qs) Parallel execution

Browse Source
This commit is contained in:
Viktor Lofgren 2023-10-23 12:06:03 +02:00
parent efb73ff4e7
commit 731afcb864
1 changed files with 10 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
code/api/index-api/src/main/java/nu/marginalia/index/client/IndexClient.java
View File

@ -8,6 +8,7 @@ import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.schedulers.Schedulers; import io.reactivex.rxjava3.schedulers.Schedulers;
import nu.marginalia.client.AbstractDynamicClient; import nu.marginalia.client.AbstractDynamicClient;
import nu.marginalia.client.Context; import nu.marginalia.client.Context;
import nu.marginalia.client.exception.RouteNotConfiguredException;
import nu.marginalia.index.client.model.query.SearchSpecification; import nu.marginalia.index.client.model.query.SearchSpecification;
import nu.marginalia.index.client.model.results.SearchResultSet; import nu.marginalia.index.client.model.results.SearchResultSet;
import nu.marginalia.model.gson.GsonFactory; import nu.marginalia.model.gson.GsonFactory;
@ -56,12 +57,15 @@ public class IndexClient extends AbstractDynamicClient {
@CheckReturnValue @CheckReturnValue
public SearchResultSet query(Context ctx, List<Integer> nodes, SearchSpecification specs) { public SearchResultSet query(Context ctx, List<Integer> nodes, SearchSpecification specs) {
return Observable.fromIterable(nodes) return Observable.fromIterable(nodes)
.concatMap(node -> this .flatMap(node -> {
.postGet(ctx, node,"/search/", specs, SearchResultSet.class) try {
.onErrorReturn(t -> new SearchResultSet()), return this
nodes.size(), .postGet(ctx, node, "/search/", specs, SearchResultSet.class).onErrorReturn(t -> new SearchResultSet())
Schedulers.io() .observeOn(Schedulers.io());
) } catch (RouteNotConfiguredException ex) {
return Observable.error(ex);
}
})
.reduce(SearchResultSet::combine) .reduce(SearchResultSet::combine)
.blockingGet(); .blockingGet();
} }