import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
public void testSimpleMultiSearchMoreRequests() {
    createIndex("test");
    int numDocs = randomIntBetween(0, 16);
    for (int i = 0; i < numDocs; i++) {
        client().prepareIndex("test", "type", Integer.toString(i)).setSource("{}", XContentType.JSON).get();
    }
    refresh();

    int numSearchRequests = randomIntBetween(1, 64);
    MultiSearchRequest request = new MultiSearchRequest();
    if (randomBoolean()) {
        request.maxConcurrentSearchRequests(randomIntBetween(1, numSearchRequests));
    }
    for (int i = 0; i < numSearchRequests; i++) {
        request.add(client().prepareSearch("test"));
    }

    MultiSearchResponse response = client().multiSearch(request).actionGet();
    assertThat(response.getResponses().length, equalTo(numSearchRequests));
    for (MultiSearchResponse.Item item : response) {
        assertNoFailures(item.getResponse());
        assertHitCount(item.getResponse(), numDocs);
    }
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
/**
 * Parses a {@link RestRequest} body and returns a {@link MultiSearchRequest}
 */
public static MultiSearchRequest parseRequest(RestRequest restRequest, boolean allowExplicitIndex) throws IOException {
    MultiSearchRequest multiRequest = new MultiSearchRequest();
    if (restRequest.hasParam("max_concurrent_searches")) {
        multiRequest.maxConcurrentSearchRequests(restRequest.paramAsInt("max_concurrent_searches", 0));
    }

    parseMultiLineRequest(restRequest, multiRequest.indicesOptions(), allowExplicitIndex, (searchRequest, parser) -> {
        try {
            final QueryParseContext queryParseContext = new QueryParseContext(parser);
            searchRequest.source(SearchSourceBuilder.fromXContent(queryParseContext));
            multiRequest.add(searchRequest);
        } catch (IOException e) {
            throw new ElasticsearchParseException("Exception when parsing search request", e);
        }
    });

    return multiRequest;
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
protected MultiSearchRequest toMultiSearchRequest(final MultiGetRequest multiGetRequest) {

        final MultiSearchRequest msearch = new MultiSearchRequest();
        msearch.copyContextFrom(multiGetRequest);

        for (final Iterator<Item> iterator = multiGetRequest.iterator(); iterator.hasNext();) {
            final Item item = iterator.next();

            final SearchRequest st = new SearchRequest();
            st.routing(item.routing());
            st.indices(item.indices());
            st.types(item.type());
            st.preference(multiGetRequest.preference());
            st.source(SearchSourceBuilder.searchSource().query(new IdsQueryBuilder(item.type()).addIds(item.id())));
            msearch.add(st);
        }

        return msearch;

    }
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
@Test
public void testMultiSearchRequestBuilder() throws IOException {
    SuggestionsRestService tested = new SuggestionsRestService();
    tested.searchClientService = Mockito.mock(SearchClientService.class);
    tested.log = Logger.getLogger("testlogger");

    MultiSearchRequestBuilder msrb = new MultiSearchRequestBuilder(null);
    SearchRequestBuilder srbNGram = new SearchRequestBuilder(null);
    SearchRequestBuilder srbFuzzy = new SearchRequestBuilder(null);

    msrb = tested.getProjectMultiSearchRequestBuilder(msrb,
            tested.getProjectSearchNGramRequestBuilder(srbNGram, "JBoss", 5),
            tested.getProjectSearchFuzzyRequestBuilder(srbFuzzy, "JBoss", 5));

    MultiSearchRequest msr = msrb.request();

    Assert.assertEquals(2, msr.requests().size());
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) throws Exception {
    MultiSearchRequest multiSearchRequest = new MultiSearchRequest();

    String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
    String[] types = Strings.splitStringByCommaToArray(request.param("type"));
    String path = request.path();
    boolean isTemplateRequest = isTemplateRequest(path);
    IndicesOptions indicesOptions = IndicesOptions.fromRequest(request, multiSearchRequest.indicesOptions());
    multiSearchRequest.add(RestActions.getRestContent(request), isTemplateRequest, indices, types, request.param("search_type"), request.param("routing"), indicesOptions, allowExplicitIndex);

    client.multiSearch(multiSearchRequest, new RestToXContentListener<MultiSearchResponse>(channel));
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
@Converter
public static MultiSearchRequest toMultiSearchRequest(Object document, Exchange exchange) {
    List<SearchRequest> items = (List<SearchRequest>) document;
    MultiSearchRequest multiSearchRequest = new MultiSearchRequest();
    Iterator<SearchRequest> it = items.iterator();
    while (it.hasNext()) {
        SearchRequest item = it.next();
        multiSearchRequest.add(item);
    }
    return multiSearchRequest;
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) throws Exception {
  MultiSearchRequest multiSearchRequest = new MultiSearchRequest();

  String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
  String[] types = Strings.splitStringByCommaToArray(request.param("type"));
  String path = request.path();
  boolean isTemplateRequest = isTemplateRequest(path);
  IndicesOptions indicesOptions = IndicesOptions.fromRequest(request, multiSearchRequest.indicesOptions());
  multiSearchRequest.add(RestActions.getRestContent(request), isTemplateRequest, indices, types, request.param("search_type"), request.param("routing"), indicesOptions, allowExplicitIndex);

  client.execute(CoordinateMultiSearchAction.INSTANCE, multiSearchRequest, new RestToXContentListener<MultiSearchResponse>(channel));
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
protected MultiSearchRequest toMultiSearchRequest(final RestRequest request) throws Exception {

        final MultiGetRequest multiGetRequest = new MultiGetRequest();
        multiGetRequest.refresh(request.paramAsBoolean("refresh", multiGetRequest.refresh()));
        multiGetRequest.preference(request.param("preference"));
        multiGetRequest.realtime(request.paramAsBoolean("realtime", null));
        multiGetRequest.ignoreErrorsOnGeneratedFields(request.paramAsBoolean("ignore_errors_on_generated_fields", false));

        String[] sFields = null;
        final String sField = request.param("fields");
        if (sField != null) {
            sFields = Strings.splitStringByCommaToArray(sField);
        }

        final FetchSourceContext defaultFetchSource = FetchSourceContext.parseFromRestRequest(request);
        multiGetRequest.add(request.param("index"), request.param("type"), sFields, defaultFetchSource, request.param("routing"),
                RestActions.getRestContent(request),
                settings.getAsBoolean(AbstractACRestFilter.REST_ACTION_MULTI_ALLOW_EXPLICIT_INDEX, true));

        final MultiSearchRequest msearch = new MultiSearchRequest();
        msearch.copyContextFrom(request);

        for (final Iterator<Item> iterator = multiGetRequest.iterator(); iterator.hasNext();) {
            final Item item = iterator.next();

            final SearchRequest st = new SearchRequest();
            st.routing(item.routing());
            st.indices(item.indices());
            st.types(item.type());
            st.preference(request.param("preference"));
            st.source(SearchSourceBuilder.searchSource().query(new IdsQueryBuilder(item.type()).addIds(item.id())));
            msearch.add(st);
        }

        return msearch;

    }
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
public void execute(MultiSearchRequest request, final ActionListener<MultiSearchResponse> listener) {
    logger.debug("multi search request {}", request);
    try {
        RequestUriBuilder uriBuilder = new RequestUriBuilder();
        uriBuilder.addEndpoint("_msearch");
        uriBuilder.addIndicesOptions(request.indicesOptions());

        // TODO convert lazily
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        for (SearchRequest searchRequest : request.requests()) {
            writeHeader(searchRequest, outputStream);
            outputStream.write(LINE_FEED);
            outputStream.write(searchRequest.source().toBytes());
            outputStream.write(LINE_FEED);
        }

        httpClient.getHttpClient().submit(HttpClientRequest.createPost(uriBuilder.toString())
                .withContent(outputStream.toByteArray()))
                .flatMap(ErrorHandler.AS_FUNC)
                .flatMap(new Func1<HttpClientResponse<ByteBuf>, Observable<MultiSearchResponse>>() {
                    @Override
                    public Observable<MultiSearchResponse> call(HttpClientResponse<ByteBuf> response) {
                        return response.getContent().flatMap(new Func1<ByteBuf, Observable<MultiSearchResponse>>() {
                            @Override
                            public Observable<MultiSearchResponse> call(ByteBuf byteBuf) {
                                return MultiSearchResponse.parse(byteBuf);
                            }
                        });
                    }
                })
                .single()
                .subscribe(new ListenerCompleterObserver<>(listener));
    } catch (Exception e) {
        listener.onFailure(e);
    }
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
@Test
public void should_do_multisearch() throws Exception {
    createSimpleDoc(THE_INDEX, THE_TYPE, "1");
    createSimpleDoc(THE_INDEX, THE_TYPE, "2");
    createSimpleDoc(THE_INDEX, THE_TYPE, "3");

    refresh();

    MultiSearchRequest request = new MultiSearchRequest()
            .add(new SearchRequest().indices(THE_INDEX).source(new SearchSourceBuilder().query(matchAllQuery())).searchType(COUNT))
            .add(new SearchRequest().indices(THE_INDEX).source(new SearchSourceBuilder().query(matchQuery("the_string_field", "the_string_value"))))
            .add(new SearchRequest().indices(THE_INDEX).source(new SearchSourceBuilder().query(matchQuery("the_string_field", "the_string_value")).from(1).size(1)))
            ;

    MultiSearchResponse response = httpClient.multiSearch(request).get();

    List<SearchResponse> responses = response.getResponses();
    Assertions.assertThat(responses).hasSize(3);

    Assertions.assertThat(responses.get(0).getHits().getTotal()).isEqualTo(3);
    Assertions.assertThat(responses.get(0).getHits().getHits()).isEmpty();

    Assertions.assertThat(responses.get(1).getHits().getTotal()).isEqualTo(3);
    Assertions.assertThat(responses.get(1).getHits().getHits()).hasSize(3);

    Assertions.assertThat(responses.get(2).getHits().getTotal()).isEqualTo(3);
    Assertions.assertThat(responses.get(2).getHits().getHits()).hasSize(1);
    Assertions.assertThat(responses.get(2).getHits().getHits().get(0).getSource().length).isGreaterThan(1);
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
@Override
public ActionFuture<MultiSearchResponse> multiSearch(MultiSearchRequest request) {
    return execute(MultiSearchAction.INSTANCE, request);
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    MultiSearchRequest multiSearchRequest = parseRequest(request, allowExplicitIndex);
    return channel -> client.multiSearch(multiSearchRequest, new RestToXContentListener<>(channel));
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
private void buildMulti() {
    multi = new MultiSearchRequest();
    multi.add(firstTable.getRequestBuilder());
    multi.add(secondTable.getRequestBuilder());
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
public MultiSearchRequest getMulti() {
    return multi;
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
public void setMulti(MultiSearchRequest multi) {
    this.multi = multi;
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
@Override
public ActionFuture<MultiSearchResponse> multiSearch(MultiSearchRequest request) {
	// TODO Auto-generated method stub
	return null;
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
private String resolveOperation(Exchange exchange) {
    // 1. Operation can be driven by either (in order of preference):
    // a. If the body is an ActionRequest the operation is set by the type
    // of request.
    // b. If the body is not an ActionRequest, the operation is set by the
    // header if it exists.
    // c. If neither the operation can not be derived from the body or
    // header, the configuration is used.
    // In the event we can't discover the operation from a, b or c we throw
    // an error.
    Object request = exchange.getIn().getBody();
    if (request instanceof IndexRequest) {
        return ElasticsearchConstants.OPERATION_INDEX;
    } else if (request instanceof GetRequest) {
        return ElasticsearchConstants.OPERATION_GET_BY_ID;
    } else if (request instanceof MultiGetRequest) {
        return ElasticsearchConstants.OPERATION_MULTIGET;
    } else if (request instanceof UpdateRequest) {
        return ElasticsearchConstants.OPERATION_UPDATE;
    } else if (request instanceof BulkRequest) {
        // do we want bulk or bulk_index?
        if ("BULK_INDEX".equals(getEndpoint().getConfig().getOperation())) {
            return ElasticsearchConstants.OPERATION_BULK_INDEX;
        } else {
            return ElasticsearchConstants.OPERATION_BULK;
        }
    } else if (request instanceof DeleteRequest) {
        return ElasticsearchConstants.OPERATION_DELETE;
    } else if (request instanceof ExistsRequest) {
        return ElasticsearchConstants.OPERATION_EXISTS;
    } else if (request instanceof SearchRequest) {
        return ElasticsearchConstants.OPERATION_SEARCH;
    } else if (request instanceof MultiSearchRequest) {
        return ElasticsearchConstants.OPERATION_MULTISEARCH;
    }

    String operationConfig = exchange.getIn().getHeader(ElasticsearchConstants.PARAM_OPERATION, String.class);
    if (operationConfig == null) {
        operationConfig = getEndpoint().getConfig().getOperation();
    }
    if (operationConfig == null) {
        throw new IllegalArgumentException(ElasticsearchConstants.PARAM_OPERATION + " value '" + operationConfig + "' is not supported");
    }
    return operationConfig;
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
public void multiSearch(MultiSearchRequest request, ActionListener<MultiSearchResponse> listener) {
    multiSearchActionHandler.execute(request, listener);
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
@Override
protected void doExecute(final MultiSearchScriptTemplateRequest request,
        final ActionListener<MultiSearchScriptTemplateResponse> listener) {
    final List<Integer> originalSlots = new ArrayList<>();
    final MultiSearchRequest multiSearchRequest = new MultiSearchRequest();
    multiSearchRequest.indicesOptions(request.indicesOptions());
    if (request.maxConcurrentSearchRequests() != 0) {
        multiSearchRequest.maxConcurrentSearchRequests(request.maxConcurrentSearchRequests());
    }

    final MultiSearchScriptTemplateResponse.Item[] items = new MultiSearchScriptTemplateResponse.Item[request.requests().size()];
    for (int i = 0; i < items.length; i++) {
        final SearchScriptTemplateRequest searchTemplateRequest = request.requests().get(i);
        final SearchScriptTemplateResponse searchTemplateResponse = new SearchScriptTemplateResponse();
        SearchRequest searchRequest;
        try {
            searchRequest = convert(searchTemplateRequest, searchTemplateResponse, scriptService, xContentRegistry);
        } catch (final Exception e) {
            items[i] = new MultiSearchScriptTemplateResponse.Item(null, e);
            continue;
        }
        items[i] = new MultiSearchScriptTemplateResponse.Item(searchTemplateResponse, null);
        if (searchRequest != null) {
            multiSearchRequest.add(searchRequest);
            originalSlots.add(i);
        }
    }

    multiSearchAction.execute(multiSearchRequest, ActionListener.wrap(r -> {
        for (int i = 0; i < r.getResponses().length; i++) {
            final MultiSearchResponse.Item item = r.getResponses()[i];
            final int originalSlot = originalSlots.get(i);
            if (item.isFailure()) {
                items[originalSlot] = new MultiSearchScriptTemplateResponse.Item(null, item.getFailure());
            } else {
                items[originalSlot].getResponse().setResponse(item.getResponse());
            }
        }
        listener.onResponse(new MultiSearchScriptTemplateResponse(items));
    }, listener::onFailure));
}
 

import org.elasticsearch.action.search.MultiSearchRequest; //导入依赖的package包/类
public MultiSearchRequestBuilder(Client client, JsonToString<JsonInput> jsonToString, StringToJson<JsonOutput> stringToJson) {
    super(client, new MultiSearchRequest(), jsonToString, stringToJson);
}