import org.elasticsearch.client.node.NodeClient; //导入依赖的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.client.node.NodeClient; //导入依赖的package包/类
public static String sqlToEsQuery(String sql) throws Exception {
        Map actions = new HashMap();
        Settings settings = Settings.builder().build();
//        Client client = new NodeClient(settings, null, null, actions);
//        Settings.builder()
//                .put(ThreadContext.PREFIX + ".key1", "val1")
//                .put(ThreadContext.PREFIX + ".key2", "val 2")
//                .build();

        ThreadPool threadPool = new ThreadPool(settings);
        Client client = new NodeClient(settings, threadPool);
        SearchDao searchDao = new org.nlpcn.es4sql.SearchDao(client);
        try {
            return searchDao.explain(sql).explain().explain();
        } catch (Exception e) {
            throw e;
        }
    }
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
    // Creates the render template request
    SearchTemplateRequest renderRequest;
    try (XContentParser parser = request.contentOrSourceParamParser()) {
        renderRequest = RestSearchTemplateAction.parse(parser);
    }
    renderRequest.setSimulate(true);

    String id = request.param("id");
    if (id != null) {
        renderRequest.setScriptType(ScriptType.STORED);
        renderRequest.setScript(id);
    }

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

import org.elasticsearch.client.node.NodeClient; //导入依赖的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.client.node.NodeClient; //导入依赖的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");
    String fieldsParam = request.param("fields");
    String defaultPipeline = request.param("pipeline");
    String[] defaultFields = fieldsParam != null ? Strings.commaDelimitedListToStringArray(fieldsParam) : null;

    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, null, defaultPipeline, null, true,
        request.getXContentType());

    // short circuit the call to the transport layer
    return channel -> {
        BulkRestBuilderListener listener = new BulkRestBuilderListener(channel, request);
        listener.onResponse(bulkRequest);
    };
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public final void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
    // prepare the request for execution; has the side effect of touching the request parameters
    final RestChannelConsumer action = prepareRequest(request, client);

    // validate unconsumed params, but we must exclude params used to format the response
    // use a sorted set so the unconsumed parameters appear in a reliable sorted order
    final SortedSet<String> unconsumedParams =
        request.unconsumedParams().stream().filter(p -> !responseParams().contains(p)).collect(Collectors.toCollection(TreeSet::new));

    // validate the non-response params
    if (!unconsumedParams.isEmpty()) {
        final Set<String> candidateParams = new HashSet<>();
        candidateParams.addAll(request.consumedParams());
        candidateParams.addAll(responseParams());
        throw new IllegalArgumentException(unrecognized(request, unconsumedParams, candidateParams, "parameter"));
    }

    // execute the action
    action.accept(channel);
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    CreateIndexRequest createIndexRequest = new CreateIndexRequest(request.param("index"));
    if (request.hasContent()) {
        createIndexRequest.source(request.content(), request.getXContentType());
    }
    createIndexRequest.updateAllTypes(request.paramAsBoolean("update_all_types", false));
    createIndexRequest.timeout(request.paramAsTime("timeout", createIndexRequest.timeout()));
    createIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", createIndexRequest.masterNodeTimeout()));
    createIndexRequest.waitForActiveShards(ActiveShardCount.parseString(request.param("wait_for_active_shards")));
    return channel -> client.admin().indices().create(createIndexRequest, new AcknowledgedRestListener<CreateIndexResponse>(channel) {
        @Override
        public void addCustomFields(XContentBuilder builder, CreateIndexResponse response) throws IOException {
            response.addCustomFields(builder);
        }
    });
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    FlushRequest flushRequest = new FlushRequest(Strings.splitStringByCommaToArray(request.param("index")));
    flushRequest.indicesOptions(IndicesOptions.fromRequest(request, flushRequest.indicesOptions()));
    flushRequest.force(request.paramAsBoolean("force", flushRequest.force()));
    flushRequest.waitIfOngoing(request.paramAsBoolean("wait_if_ongoing", flushRequest.waitIfOngoing()));
    return channel -> client.admin().indices().flush(flushRequest, new RestBuilderListener<FlushResponse>(channel) {
        @Override
        public RestResponse buildResponse(FlushResponse response, XContentBuilder builder) throws Exception {
            builder.startObject();
            buildBroadcastShardsHeader(builder, request, response);
            builder.endObject();
            return new BytesRestResponse(OK, builder);
        }
    });
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    PutIndexTemplateRequest putRequest = new PutIndexTemplateRequest(request.param("name"));
    if (request.hasParam("template")) {
        DEPRECATION_LOGGER.deprecated("Deprecated parameter[template] used, replaced by [index_patterns]");
        putRequest.patterns(Collections.singletonList(request.param("template")));
    } else {
        putRequest.patterns(Arrays.asList(request.paramAsStringArray("index_patterns", Strings.EMPTY_ARRAY)));
    }
    putRequest.order(request.paramAsInt("order", putRequest.order()));
    putRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putRequest.masterNodeTimeout()));
    putRequest.create(request.paramAsBoolean("create", false));
    putRequest.cause(request.param("cause", ""));
    putRequest.source(request.content(), request.getXContentType());
    return channel -> client.admin().indices().putTemplate(putRequest, new AcknowledgedRestListener<>(channel));
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
public InternalClusterInfoService(Settings settings, ClusterService clusterService, ThreadPool threadPool, NodeClient client) {
    super(settings);
    this.leastAvailableSpaceUsages = ImmutableOpenMap.of();
    this.mostAvailableSpaceUsages = ImmutableOpenMap.of();
    this.shardRoutingToDataPath = ImmutableOpenMap.of();
    this.shardSizes = ImmutableOpenMap.of();
    this.clusterService = clusterService;
    this.threadPool = threadPool;
    this.client = client;
    this.updateFrequency = INTERNAL_CLUSTER_INFO_UPDATE_INTERVAL_SETTING.get(settings);
    this.fetchTimeout = INTERNAL_CLUSTER_INFO_TIMEOUT_SETTING.get(settings);
    this.enabled = DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_THRESHOLD_ENABLED_SETTING.get(settings);
    ClusterSettings clusterSettings = clusterService.getClusterSettings();
    clusterSettings.addSettingsUpdateConsumer(INTERNAL_CLUSTER_INFO_TIMEOUT_SETTING, this::setFetchTimeout);
    clusterSettings.addSettingsUpdateConsumer(INTERNAL_CLUSTER_INFO_UPDATE_INTERVAL_SETTING, this::setUpdateFrequency);
    clusterSettings.addSettingsUpdateConsumer(DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_THRESHOLD_ENABLED_SETTING, this::setEnabled);

    // Add InternalClusterInfoService to listen for Master changes
    this.clusterService.addLocalNodeMasterListener(this);
    // Add to listen for state changes (when nodes are added)
    this.clusterService.addListener(this);
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    final String[] names = Strings.splitStringByCommaToArray(request.param("name"));

    final GetIndexTemplatesRequest getIndexTemplatesRequest = new GetIndexTemplatesRequest(names);
    getIndexTemplatesRequest.local(request.paramAsBoolean("local", getIndexTemplatesRequest.local()));
    getIndexTemplatesRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getIndexTemplatesRequest.masterNodeTimeout()));

    final boolean implicitAll = getIndexTemplatesRequest.names().length == 0;

    return channel ->
            client.admin()
                    .indices()
                    .getTemplates(getIndexTemplatesRequest, new RestToXContentListener<GetIndexTemplatesResponse>(channel) {
                        @Override
                        protected RestStatus getStatus(final GetIndexTemplatesResponse response) {
                            final boolean templateExists = response.getIndexTemplates().isEmpty() == false;
                            return (templateExists || implicitAll) ? OK : NOT_FOUND;
                        }
                    });
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    ForceMergeRequest mergeRequest = new ForceMergeRequest(Strings.splitStringByCommaToArray(request.param("index")));
    mergeRequest.indicesOptions(IndicesOptions.fromRequest(request, mergeRequest.indicesOptions()));
    mergeRequest.maxNumSegments(request.paramAsInt("max_num_segments", mergeRequest.maxNumSegments()));
    mergeRequest.onlyExpungeDeletes(request.paramAsBoolean("only_expunge_deletes", mergeRequest.onlyExpungeDeletes()));
    mergeRequest.flush(request.paramAsBoolean("flush", mergeRequest.flush()));
    return channel -> client.admin().indices().forceMerge(mergeRequest, new RestBuilderListener<ForceMergeResponse>(channel) {
        @Override
        public RestResponse buildResponse(ForceMergeResponse response, XContentBuilder builder) throws Exception {
            builder.startObject();
            buildBroadcastShardsHeader(builder, request, response);
            builder.endObject();
            return new BytesRestResponse(OK, builder);
        }
    });
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    if (request.param("target") == null) {
        throw new IllegalArgumentException("no target index");
    }
    if (request.param("index") == null) {
        throw new IllegalArgumentException("no source index");
    }
    ShrinkRequest shrinkIndexRequest = new ShrinkRequest(request.param("target"), request.param("index"));
    request.applyContentParser(parser -> ShrinkRequest.PARSER.parse(parser, shrinkIndexRequest, null));
    shrinkIndexRequest.timeout(request.paramAsTime("timeout", shrinkIndexRequest.timeout()));
    shrinkIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", shrinkIndexRequest.masterNodeTimeout()));
    shrinkIndexRequest.setWaitForActiveShards(ActiveShardCount.parseString(request.param("wait_for_active_shards")));
    return channel -> client.admin().indices().shrinkIndex(shrinkIndexRequest, new AcknowledgedRestListener<ShrinkResponse>(channel) {
        @Override
        public void addCustomFields(XContentBuilder builder, ShrinkResponse response) throws IOException {
            response.addCustomFields(builder);
        }
    });
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {

    final RecoveryRequest recoveryRequest = new RecoveryRequest(Strings.splitStringByCommaToArray(request.param("index")));
    recoveryRequest.detailed(request.paramAsBoolean("detailed", false));
    recoveryRequest.activeOnly(request.paramAsBoolean("active_only", false));
    recoveryRequest.indicesOptions(IndicesOptions.fromRequest(request, recoveryRequest.indicesOptions()));

    return channel -> client.admin().indices().recoveries(recoveryRequest, new RestBuilderListener<RecoveryResponse>(channel) {
        @Override
        public RestResponse buildResponse(RecoveryResponse response, XContentBuilder builder) throws Exception {
            response.detailed(recoveryRequest.detailed());
            builder.startObject();
            response.toXContent(builder, request);
            builder.endObject();
            return new BytesRestResponse(OK, builder);
        }
    });

}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    ClearIndicesCacheRequest clearIndicesCacheRequest = new ClearIndicesCacheRequest(
            Strings.splitStringByCommaToArray(request.param("index")));
    clearIndicesCacheRequest.indicesOptions(IndicesOptions.fromRequest(request, clearIndicesCacheRequest.indicesOptions()));
    fromRequest(request, clearIndicesCacheRequest);
    return channel ->
            client.admin().indices().clearCache(clearIndicesCacheRequest, new RestBuilderListener<ClearIndicesCacheResponse>(channel) {
        @Override
        public RestResponse buildResponse(ClearIndicesCacheResponse response, XContentBuilder builder) throws Exception {
            builder.startObject();
            buildBroadcastShardsHeader(builder, request, response);
            builder.endObject();
            return new BytesRestResponse(OK, builder);
        }
    });
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    IndicesShardStoresRequest indicesShardStoresRequest = new IndicesShardStoresRequest(
            Strings.splitStringByCommaToArray(request.param("index")));
    if (request.hasParam("status")) {
        indicesShardStoresRequest.shardStatuses(Strings.splitStringByCommaToArray(request.param("status")));
    }
    indicesShardStoresRequest.indicesOptions(IndicesOptions.fromRequest(request, indicesShardStoresRequest.indicesOptions()));
    return channel ->
        client.admin()
            .indices()
            .shardStores(indicesShardStoresRequest, new RestBuilderListener<IndicesShardStoresResponse>(channel) {
                @Override
                public RestResponse buildResponse(
                    IndicesShardStoresResponse response,
                    XContentBuilder builder) throws Exception {
                    builder.startObject();
                    response.toXContent(builder, request);
                    builder.endObject();
                    return new BytesRestResponse(OK, builder);
                }
            });
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    MultiGetRequest multiGetRequest = new MultiGetRequest();
    multiGetRequest.refresh(request.paramAsBoolean("refresh", multiGetRequest.refresh()));
    multiGetRequest.preference(request.param("preference"));
    multiGetRequest.realtime(request.paramAsBoolean("realtime", multiGetRequest.realtime()));
    if (request.param("fields") != null) {
        throw new IllegalArgumentException("The parameter [fields] is no longer supported, " +
            "please use [stored_fields] to retrieve stored fields or _source filtering if the field is not stored");
    }
    String[] sFields = null;
    String sField = request.param("stored_fields");
    if (sField != null) {
        sFields = Strings.splitStringByCommaToArray(sField);
    }

    FetchSourceContext defaultFetchSource = FetchSourceContext.parseFromRestRequest(request);
    try (XContentParser parser = request.contentOrSourceParamParser()) {
        multiGetRequest.add(request.param("index"), request.param("type"), sFields, defaultFetchSource,
            request.param("routing"), parser, allowExplicitIndex);
    }

    return channel -> client.multiGet(multiGetRequest, new RestToXContentListener<>(channel));
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
/**
 *@description 插件的业务处理方法
 *@time 创建时间:2017年8月31日下午4:21:52
 *@param request
 *@param client
 *@return
 *@throws IOException
 *@author dzn
 */
@Override
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client)
        throws IOException {
    LOGGER.info("My Rest Action Handler");  
    Map<String, String> params = request.params();
    LOGGER.info("Param : {}", params);
    final String actionParam = request.param("action");
    LOGGER.info("actionParam : {}", actionParam);
    RestChannelConsumer res = new RestChannelConsumer(){
        @Override
        public void accept(RestChannel channel) throws Exception {
            XContentBuilder builder = channel.newBuilder();
            builder.startObject();  
            //返回给页面的信息
            builder.field("message", "This is my first plugin");
            if(null != actionParam && actionParam.trim().length() > 0){
                builder.field("接收到参数actionParam", actionParam);
            }
            builder.endObject();  
            channel.sendResponse(new BytesRestResponse(RestStatus.OK, builder));
        }
    };
    return res;
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
    String id = request.param("id");
    String lang = request.param("lang");

    // In the case where only {lang} is not null, we make it {id} because of
    // name ordering issues in the handlers' paths.
    if (id == null) {
        id = lang;
        lang = null;
    }

    if (lang != null) {
        deprecationLogger.deprecated(
            "specifying lang [" + lang + "] as part of the url path is deprecated");
    }

    DeleteStoredScriptRequest deleteStoredScriptRequest = new DeleteStoredScriptRequest(id, lang);
    return channel -> client.admin().cluster().deleteStoredScript(deleteStoredScriptRequest, new AcknowledgedRestListener<>(channel));
}
 

import org.elasticsearch.client.node.NodeClient; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    final String[] nodesIds = Strings.splitStringByCommaToArray(request.param("nodes"));
    final TaskId taskId = new TaskId(request.param("task_id"));
    final String[] actions = Strings.splitStringByCommaToArray(request.param("actions"));
    final TaskId parentTaskId = new TaskId(request.param("parent_task_id"));
    final String groupBy = request.param("group_by", "nodes");

    CancelTasksRequest cancelTasksRequest = new CancelTasksRequest();
    cancelTasksRequest.setTaskId(taskId);
    cancelTasksRequest.setNodes(nodesIds);
    cancelTasksRequest.setActions(actions);
    cancelTasksRequest.setParentTaskId(parentTaskId);
    return channel ->
        client.admin().cluster().cancelTasks(cancelTasksRequest, listTasksResponseListener(nodesInCluster, groupBy, channel));
}