import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
    if (request.hasContentOrSourceParam() == false) {
        throw new ElasticsearchException("request body is required");
    }

    // Creates the search request with all required params
    SearchRequest searchRequest = new SearchRequest();
    RestSearchAction.parseSearchRequest(searchRequest, request, null);

    // Creates the search template request
    SearchTemplateRequest searchTemplateRequest;
    try (XContentParser parser = request.contentOrSourceParamParser()) {
        searchTemplateRequest = PARSER.parse(parser, new SearchTemplateRequest(), null);
    }
    searchTemplateRequest.setRequest(searchRequest);

    return channel -> client.execute(SearchTemplateAction.INSTANCE, searchTemplateRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    String scrollIds = request.param("scroll_id");
    ClearScrollRequest clearRequest = new ClearScrollRequest();
    clearRequest.setScrollIds(Arrays.asList(splitScrollIds(scrollIds)));
    request.withContentOrSourceParamParserOrNull((xContentParser -> {
        if (xContentParser != null) {
            // NOTE: if rest request with xcontent body has request parameters, these parameters does not override xcontent value
            clearRequest.setScrollIds(null);
            try {
                buildFromContent(xContentParser, clearRequest);
            } catch (IOException e) {
                throw new IllegalArgumentException("Failed to parse request body", e);
            }
        }
    }));

    return channel -> client.clearScroll(clearRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    String scrollId = request.param("scroll_id");
    SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
    searchScrollRequest.scrollId(scrollId);
    String scroll = request.param("scroll");
    if (scroll != null) {
        searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
    }

    request.withContentOrSourceParamParserOrNull(xContentParser -> {
        if (xContentParser != null) {
            // NOTE: if rest request with xcontent body has request parameters, these parameters override xcontent values
            try {
                buildFromContent(xContentParser, searchScrollRequest);
            } catch (IOException e) {
                throw new IllegalArgumentException("Failed to parse request body", e);
            }
        }});
    return channel -> client.searchScroll(searchScrollRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    DeleteRequest deleteRequest = new DeleteRequest(request.param("index"), request.param("type"), request.param("id"));
    deleteRequest.routing(request.param("routing"));
    deleteRequest.parent(request.param("parent"));
    deleteRequest.timeout(request.paramAsTime("timeout", DeleteRequest.DEFAULT_TIMEOUT));
    deleteRequest.setRefreshPolicy(request.param("refresh"));
    deleteRequest.version(RestActions.parseVersion(request));
    deleteRequest.versionType(VersionType.fromString(request.param("version_type"), deleteRequest.versionType()));

    String waitForActiveShards = request.param("wait_for_active_shards");
    if (waitForActiveShards != null) {
        deleteRequest.waitForActiveShards(ActiveShardCount.parseString(waitForActiveShards));
    }

    return channel -> client.delete(deleteRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    BulkRequest bulkRequest = Requests.bulkRequest();
    String defaultIndex = request.param("index");
    String defaultType = request.param("type");
    String defaultRouting = request.param("routing");
    FetchSourceContext defaultFetchSourceContext = FetchSourceContext.parseFromRestRequest(request);
    String fieldsParam = request.param("fields");
    if (fieldsParam != null) {
        DEPRECATION_LOGGER.deprecated("Deprecated field [fields] used, expected [_source] instead");
    }
    String[] defaultFields = fieldsParam != null ? Strings.commaDelimitedListToStringArray(fieldsParam) : null;
    String defaultPipeline = request.param("pipeline");
    String waitForActiveShards = request.param("wait_for_active_shards");
    if (waitForActiveShards != null) {
        bulkRequest.waitForActiveShards(ActiveShardCount.parseString(waitForActiveShards));
    }
    bulkRequest.timeout(request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT));
    bulkRequest.setRefreshPolicy(request.param("refresh"));
    bulkRequest.add(request.content(), defaultIndex, defaultType, defaultRouting, defaultFields,
        defaultFetchSourceContext, defaultPipeline, null, allowExplicitIndex, request.getXContentType());

    return channel -> client.bulk(bulkRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    IndexRequest indexRequest = new IndexRequest(request.param("index"), request.param("type"), request.param("id"));
    indexRequest.routing(request.param("routing"));
    indexRequest.parent(request.param("parent"));
    indexRequest.setPipeline(request.param("pipeline"));
    indexRequest.source(request.content(), request.getXContentType());
    indexRequest.timeout(request.paramAsTime("timeout", IndexRequest.DEFAULT_TIMEOUT));
    indexRequest.setRefreshPolicy(request.param("refresh"));
    indexRequest.version(RestActions.parseVersion(request));
    indexRequest.versionType(VersionType.fromString(request.param("version_type"), indexRequest.versionType()));
    String sOpType = request.param("op_type");
    String waitForActiveShards = request.param("wait_for_active_shards");
    if (waitForActiveShards != null) {
        indexRequest.waitForActiveShards(ActiveShardCount.parseString(waitForActiveShards));
    }
    if (sOpType != null) {
        indexRequest.opType(sOpType);
    }

    return channel ->
            client.index(indexRequest, new RestStatusToXContentListener<>(channel, r -> r.getLocation(indexRequest.routing())));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
/**
 * Execute the ActionRequest and returns the REST response using the channel.
 */
public void execute() throws Exception {
       ActionRequest request = requestBuilder.request();

       //todo: maby change to instanceof multi?
       if(requestBuilder instanceof JoinRequestBuilder){
           executeJoinRequestAndSendResponse();
       }
	else if (request instanceof SearchRequest) {
		client.search((SearchRequest) request, new RestStatusToXContentListener<SearchResponse>(channel));
	} else if (requestBuilder instanceof SqlElasticDeleteByQueryRequestBuilder) {
           throw new UnsupportedOperationException("currently not support delete on elastic 2.0.0");
       }
       else if(request instanceof GetIndexRequest) {
           this.requestBuilder.getBuilder().execute( new GetIndexRequestRestListener(channel, (GetIndexRequest) request));
       }


	else {
		throw new Exception(String.format("Unsupported ActionRequest provided: %s", request.getClass().getName()));
	}
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    // Creates the search request with all required params
    final SearchRequest searchRequest = new SearchRequest();
    RestSearchAction.parseSearchRequest(searchRequest, request, null);

    // Creates the search template request
    SearchScriptTemplateRequest searchTemplateRequest;
    try (XContentParser parser = request.contentOrSourceParamParser()) {
        searchTemplateRequest = PARSER.parse(parser, new SearchScriptTemplateRequest(), null);
    }
    searchTemplateRequest.setRequest(searchRequest);

    return channel -> client.execute(SearchScriptTemplateAction.INSTANCE, searchTemplateRequest,
            new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
    final LangdetectRequest langdetectRequest = new LangdetectRequest();
    langdetectRequest.setText(request.param("text"));
    langdetectRequest.setProfile(request.param("profile", ""));
    withContent(request, parser -> {
        if (parser != null) {
            XContentParser.Token token;
            while ((token = parser.nextToken()) != null) {
                if (token == XContentParser.Token.VALUE_STRING) {
                    if ("text".equals(parser.currentName())) {
                        langdetectRequest.setText(parser.text());
                    } else if ("profile".equals(parser.currentName())) {
                        langdetectRequest.setProfile(parser.text());
                    }
                }
            }
        }
    });
    return channel -> client.execute(LangdetectAction.INSTANCE, langdetectRequest,
            new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    ClusterHealthRequest clusterHealthRequest = clusterHealthRequest(Strings.splitStringByCommaToArray(request.param("index")));
    clusterHealthRequest.local(request.paramAsBoolean("local", clusterHealthRequest.local()));
    clusterHealthRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterHealthRequest.masterNodeTimeout()));
    clusterHealthRequest.timeout(request.paramAsTime("timeout", clusterHealthRequest.timeout()));
    String waitForStatus = request.param("wait_for_status");
    if (waitForStatus != null) {
        clusterHealthRequest.waitForStatus(ClusterHealthStatus.valueOf(waitForStatus.toUpperCase(Locale.ROOT)));
    }
    clusterHealthRequest.waitForNoRelocatingShards(
            request.paramAsBoolean("wait_for_no_relocating_shards", clusterHealthRequest.waitForNoRelocatingShards()));
    if (request.hasParam("wait_for_relocating_shards")) {
        // wait_for_relocating_shards has been removed in favor of wait_for_no_relocating_shards
        throw new IllegalArgumentException("wait_for_relocating_shards has been removed, " +
                                           "use wait_for_no_relocating_shards [true/false] instead");
    }
    String waitForActiveShards = request.param("wait_for_active_shards");
    if (waitForActiveShards != null) {
        clusterHealthRequest.waitForActiveShards(ActiveShardCount.parseString(waitForActiveShards));
    }
    clusterHealthRequest.waitForNodes(request.param("wait_for_nodes", clusterHealthRequest.waitForNodes()));
    if (request.param("wait_for_events") != null) {
        clusterHealthRequest.waitForEvents(Priority.valueOf(request.param("wait_for_events").toUpperCase(Locale.ROOT)));
    }
    return channel -> client.admin().cluster().health(clusterHealthRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    SearchRequest searchRequest = new SearchRequest();
    request.withContentOrSourceParamParserOrNull(parser ->
        parseSearchRequest(searchRequest, request, parser));

    return channel -> client.search(searchRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
RestChannelConsumer search(NodeClient client, String type, String indexName, RestRequest request) {
    String prefix = request.param("prefix");
    int from = request.paramAsInt("from", 0);
    int size = request.paramAsInt("size", 20);
    BoolQueryBuilder qb = boolQuery().filter(termQuery("type", type));
    if (prefix != null && !prefix.isEmpty()) {
        qb.must(matchQuery("name.prefix", prefix));
    }
    return (channel) -> client.prepareSearch(indexName)
            .setTypes(IndexFeatureStore.ES_TYPE)
            .setQuery(qb)
            .setSize(size)
            .setFrom(from)
            .execute(new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
    String store = indexName(request);
    String setName = request.param("name");
    String routing = request.param("routing");
    String featureQuery = null;
    List<StoredFeature> features = null;
    boolean merge = request.paramAsBoolean("merge", false);
    if (request.hasParam("query")) {
        featureQuery = request.param("query");
    }
    FeatureValidation validation = null;
    if (request.hasContentOrSourceParam()) {
        FeaturesParserState featuresParser = new FeaturesParserState();
        request.applyContentParser(featuresParser::parse);
        features = featuresParser.features;
        validation = featuresParser.validation;
    }
    if (featureQuery == null && (features == null || features.isEmpty())) {
        throw new IllegalArgumentException("features must be provided as a query for the feature store " +
                "or in the body, none provided");
    }

    if (featureQuery != null && (features != null && !features.isEmpty())) {
        throw new IllegalArgumentException("features must be provided as a query for the feature store " +
                "or directly in the body not both");
    }

    AddFeaturesToSetRequestBuilder builder = AddFeaturesToSetAction.INSTANCE.newRequestBuilder(client);
    builder.request().setStore(store);
    builder.request().setFeatureSet(setName);
    builder.request().setFeatureNameQuery(featureQuery);
    builder.request().setRouting(routing);
    builder.request().setFeatures(features);
    builder.request().setMerge(merge);
    builder.request().setValidation(validation);
    return (channel) -> builder.execute(new RestStatusToXContentListener<>(channel, (r) -> r.getResponse().getLocation(routing)));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
    final String value = request.param("value");
    final ISBNFormatRequest isbnFormatRequest = new ISBNFormatRequest().setValue(value);
    return channel -> client.execute(ISBNFormatAction.INSTANCE, isbnFormatRequest,
                new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    SearchRequest searchRequest = new SearchRequest();
    return channel -> client.execute(NoopSearchAction.INSTANCE, searchRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
    GetPipelineRequest request = new GetPipelineRequest(Strings.splitStringByCommaToArray(restRequest.param("id")));
    request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
    return channel -> client.admin().cluster().getPipeline(request, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    UpdateRequest updateRequest = new UpdateRequest(request.param("index"), request.param("type"), request.param("id"));
    updateRequest.routing(request.param("routing"));
    updateRequest.parent(request.param("parent"));
    updateRequest.timeout(request.paramAsTime("timeout", updateRequest.timeout()));
    updateRequest.setRefreshPolicy(request.param("refresh"));
    String waitForActiveShards = request.param("wait_for_active_shards");
    if (waitForActiveShards != null) {
        updateRequest.waitForActiveShards(ActiveShardCount.parseString(waitForActiveShards));
    }
    updateRequest.docAsUpsert(request.paramAsBoolean("doc_as_upsert", updateRequest.docAsUpsert()));
    FetchSourceContext fetchSourceContext = FetchSourceContext.parseFromRestRequest(request);
    String sField = request.param("fields");
    if (sField != null && fetchSourceContext != null) {
        throw new IllegalArgumentException("[fields] and [_source] cannot be used in the same request");
    }
    if (sField != null) {
        DEPRECATION_LOGGER.deprecated("Deprecated field [fields] used, expected [_source] instead");
        String[] sFields = Strings.splitStringByCommaToArray(sField);
        updateRequest.fields(sFields);
    } else if (fetchSourceContext != null) {
        updateRequest.fetchSource(fetchSourceContext);
    }

    updateRequest.retryOnConflict(request.paramAsInt("retry_on_conflict", updateRequest.retryOnConflict()));
    updateRequest.version(RestActions.parseVersion(request));
    updateRequest.versionType(VersionType.fromString(request.param("version_type"), updateRequest.versionType()));


    request.applyContentParser(parser -> {
        updateRequest.fromXContent(parser);
        IndexRequest upsertRequest = updateRequest.upsertRequest();
        if (upsertRequest != null) {
            upsertRequest.routing(request.param("routing"));
            upsertRequest.parent(request.param("parent"));
            upsertRequest.version(RestActions.parseVersion(request));
            upsertRequest.versionType(VersionType.fromString(request.param("version_type"), upsertRequest.versionType()));
        }
        IndexRequest doc = updateRequest.doc();
        if (doc != null) {
            doc.routing(request.param("routing"));
            doc.parent(request.param("parent")); // order is important, set it after routing, so it will set the routing
            doc.version(RestActions.parseVersion(request));
            doc.versionType(VersionType.fromString(request.param("version_type"), doc.versionType()));
        }
    });

    return channel ->
            client.update(updateRequest, new RestStatusToXContentListener<>(channel, r -> r.getLocation(updateRequest.routing())));
}
 

import org.elasticsearch.rest.action.RestStatusToXContentListener; //导入依赖的package包/类
@Override
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
    String store = indexName(request);
    Long expectedVersion = null;
    if (request.hasParam("version")) {
        expectedVersion = request.paramAsLong("version", -1);
        if (expectedVersion <= 0) {
            throw new IllegalArgumentException("version must be a strictly positive long value");
        }
    }
    String routing = request.param("routing");
    ParserState state = new ParserState();
    request.withContentOrSourceParamParserOrNull((p) -> ParserState.parse(p, state));
    CreateModelFromSetRequestBuilder builder = CreateModelFromSetAction.INSTANCE.newRequestBuilder(client);
    if (expectedVersion != null) {
        builder.withVersion(store, request.param("name"), expectedVersion, state.model.name, state.model.model);
    } else {
        builder.withoutVersion(store, request.param("name"), state.model.name, state.model.model);
    }
    builder.request().setValidation(state.validation);
    builder.routing(routing);
    return (channel) -> builder.execute(ActionListener.wrap(
            response -> new RestStatusToXContentListener<CreateModelFromSetAction.CreateModelFromSetResponse>(channel,
                    (r) -> r.getResponse().getLocation(routing)).onResponse(response),
            (e) -> {
                final Exception exc;
                final RestStatus status;
                if (ExceptionsHelper.unwrap(e, VersionConflictEngineException.class) != null) {
                    exc = new IllegalArgumentException("Element of type [" + StoredLtrModel.TYPE +
                            "] are not updatable, please create a new one instead.");
                    exc.addSuppressed(e);
                    status = RestStatus.METHOD_NOT_ALLOWED;
                } else {
                    exc = e;
                    status = ExceptionsHelper.status(exc);
                }

                try {
                    channel.sendResponse(new BytesRestResponse(channel, status, exc));
                } catch (Exception inner) {
                    inner.addSuppressed(e);
                    logger.error("failed to send failure response", inner);
                }
            }
    ));
}