import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
public static void buildBroadcastShardsHeader(XContentBuilder builder, Params params,
                                              int total, int successful, int failed,
                                              ShardOperationFailedException[] shardFailures) throws IOException {
    builder.startObject("_shards");
    builder.field("total", total);
    builder.field("successful", successful);
    builder.field("failed", failed);
    if (shardFailures != null && shardFailures.length > 0) {
        builder.startArray("failures");
        final boolean group = params.paramAsBoolean("group_shard_failures", true); // we group by default
        for (ShardOperationFailedException shardFailure : group ? ExceptionsHelper.groupBy(shardFailures) : shardFailures) {
            builder.startObject();
            shardFailure.toXContent(builder, params);
            builder.endObject();
        }
        builder.endArray();
    }
    builder.endObject();
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
/**
 * Create the XContent header for any {@link BaseNodesResponse}. This looks like:
 * <code>
 * "_nodes" : {
 *   "total" : 3,
 *   "successful" : 1,
 *   "failed" : 2,
 *   "failures" : [ { ... }, { ... } ]
 * }
 * </code>
 * Prefer the overload that properly invokes this method to calling this directly.
 *
 * @param builder XContent builder.
 * @param params XContent parameters.
 * @param total The total number of nodes touched.
 * @param successful The successful number of responses received.
 * @param failed The number of failures (effectively {@code total - successful}).
 * @param failures The failure exceptions related to {@code failed}.
 * @see #buildNodesHeader(XContentBuilder, Params, BaseNodesResponse)
 */
public static void buildNodesHeader(final XContentBuilder builder, final Params params,
                                    final int total, final int successful, final int failed,
                                    final List<FailedNodeException> failures) throws IOException {
    builder.startObject("_nodes");
    builder.field("total", total);
    builder.field("successful", successful);
    builder.field("failed", failed);

    if (failures.isEmpty() == false) {
        builder.startArray("failures");
        for (FailedNodeException failure : failures) {
            builder.startObject();
            failure.toXContent(builder, params);
            builder.endObject();
        }
        builder.endArray();
    }

    builder.endObject();
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
public static Settings filterSettings(Params params, Settings settings) {
    String patterns = params.param(SETTINGS_FILTER_PARAM);
    final Settings filteredSettings;
    if (patterns != null && patterns.isEmpty() == false) {
        filteredSettings = filterSettings(Strings.commaDelimitedListToSet(patterns), settings);
    } else {
        filteredSettings = settings;
    }
    return filteredSettings;
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
/**
 * Writes a "raw" (bytes) field, handling cases where the bytes are compressed, and tries to optimize writing using
 * {@link XContentBuilder#rawField(String, org.elasticsearch.common.bytes.BytesReference)}.
 * @deprecated use {@link #writeRawField(String, BytesReference, XContentType, XContentBuilder, Params)} to avoid content type
 * auto-detection
 */
@Deprecated
public static void writeRawField(String field, BytesReference source, XContentBuilder builder, ToXContent.Params params) throws IOException {
    Compressor compressor = CompressorFactory.compressor(source);
    if (compressor != null) {
        InputStream compressedStreamInput = compressor.streamInput(source.streamInput());
        builder.rawField(field, compressedStreamInput);
    } else {
        builder.rawField(field, source);
    }
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
public static Settings filterSettings(Params params, Settings settings) {
    String patterns = params.param(SETTINGS_FILTER_PARAM);
    Settings filteredSettings = settings;
    if (patterns != null && patterns.isEmpty() == false) {
        filteredSettings = SettingsFilter.filterSettings(patterns, filteredSettings);
    }
    return filteredSettings;
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
/**
 * Writes a "raw" (bytes) field, handling cases where the bytes are compressed, and tries to optimize writing using
 * {@link XContentBuilder#rawField(String, org.elasticsearch.common.bytes.BytesReference)}.
 */
public static void writeRawField(String field, BytesReference source, XContentBuilder builder, ToXContent.Params params) throws IOException {
    Compressor compressor = CompressorFactory.compressor(source);
    if (compressor != null) {
        InputStream compressedStreamInput = compressor.streamInput(source.streamInput());
        builder.rawField(field, compressedStreamInput);
    } else {
        builder.rawField(field, source);
    }
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
void toXContent(XContentBuilder builder, Params params) throws IOException {
    builder.startObject();
    builder.field(CommonFields.KEY, term);
    builder.field(CommonFields.DOC_COUNT, count);
    aggregations.toXContentInternal(builder, params);
    builder.endObject();
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
/**
 * Execute the reindexing
 *
 * @param params   Rest request
 * @param content  Content of rest request in {}
 * @param listener is to receive the response back
 * @return
 */
public String execute(final Params params, final BytesReference content, final ActionListener<Void> listener) {

    final String url = params.param("url");
    // set scroll to 1m if there is no
    final String scroll = params.param("scroll", "1m");
    final String fromIndex = params.param("index");
    final String fromType = params.param("type");
    final String toIndex = params.param("toindex");
    final String toType = params.param("totype");
    final String[] fields = params.paramAsBoolean("parent", true) ? new String[]{"_source", "_parent"} : new String[]{"_source"};
    final boolean deletion = params.paramAsBoolean("deletion", false);

    final ReindexingListener reindexingListener = new ReindexingListener(url, fromIndex, fromType, toIndex, toType, scroll, deletion, listener);

    // Create search request builder
    final SearchRequestBuilder builder = client.prepareSearch(fromIndex)
            .setScroll(scroll).addFields(fields);
    if (fromType != null && fromType.trim().length() > 0) {
        builder.setTypes(fromType.split(","));
    }
    if (content == null || content.length() == 0) {
        builder.setQuery(QueryBuilders.matchAllQuery()).setSize(
                Integer.parseInt(params.param("size", "1000")));
    } else {
        builder.setExtraSource(content);
    }
    builder.execute(reindexingListener);  // async
    reindexingListenerMap.put(reindexingListener.getName(), reindexingListener);
    return reindexingListener.getName();
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
public void execute(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap) {
    synchronized (handlers) {
        if (position < handlers.length) {
            final RequestHandler handler = handlers[position];
            position++;
            handler.execute(params, listener, requestMap, paramMap, this);
        }
    }
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
private void doItemIndexExists(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap, final RequestHandlerChain chain) {
    final String index = params.param(
            TasteConstants.REQUEST_PARAM_ITEM_INDEX, params.param("index"));

    try {
        indexCreationLock.lock();
        final IndicesExistsResponse indicesExistsResponse = client.admin()
                .indices().prepareExists(index).execute().actionGet();
        if (indicesExistsResponse.isExists()) {
            doItemMappingCreation(params, listener, requestMap, paramMap,
                    chain);
        } else {
            doItemIndexCreation(params, listener, requestMap, paramMap,
                    chain, index);
        }
    } catch (final Exception e) {
        final List<Throwable> errorList = getErrorList(paramMap);
        if (errorList.size() >= maxRetryCount) {
            listener.onError(e);
        } else {
            sleep(e);
            errorList.add(e);
            fork(() -> execute(params, listener, requestMap, paramMap,
                    chain));
        }
    } finally {
        indexCreationLock.unlock();
    }
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
private void doItemUpdate(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap,
        final Map<String, Object> itemMap, final String index,
        final String type, final String itemIdField,
        final String timestampField, final Long itemId,
        final OpType opType, final RequestHandlerChain chain) {
    itemMap.put(itemIdField, itemId);
    itemMap.put(timestampField, new Date());
    final OnResponseListener<IndexResponse> responseListener = response -> {
        paramMap.put(itemIdField, itemId);
        chain.execute(params, listener, requestMap, paramMap);
    };
    final OnFailureListener failureListener = t -> {
        sleep(t);
        if (t instanceof DocumentAlreadyExistsException
                || t instanceof EsRejectedExecutionException) {
            execute(params, listener, requestMap, paramMap, chain);
        } else {
            listener.onError(t);
        }
    };
    client.prepareIndex(index, type, itemId.toString()).setSource(itemMap)
            .setRefresh(true).setOpType(opType)
            .execute(on(responseListener, failureListener));
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
private void doPreferenceIndexExists(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap, final RequestHandlerChain chain) {
    final String index = params.param("index");

    try {
        indexCreationLock.lock();
        final IndicesExistsResponse indicesExistsResponse = client.admin()
                .indices().prepareExists(index).execute().actionGet();
        if (indicesExistsResponse.isExists()) {
            doPreferenceMappingCreation(params, listener, requestMap,
                    paramMap, chain);
        } else {
            doPreferenceIndexCreation(params, listener, requestMap,
                    paramMap, chain, index);
        }
    } catch (final Exception e) {
        final List<Throwable> errorList = getErrorList(paramMap);
        if (errorList.size() >= maxRetryCount) {
            listener.onError(e);
        } else {
            sleep(e);
            errorList.add(e);
            fork(() -> execute(params, listener, requestMap, paramMap,
                    chain));
        }
    } finally {
        indexCreationLock.unlock();
    }
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
public MultiTermVectorsListener(final int numOfThread,
        final RequestHandler[] requestHandlers,
        final Params eventParams, final Map<String, DocInfo> idMap,
        final ExecutorService executor, final ESLogger logger) {
    this.requestHandlers = requestHandlers;
    this.eventParams = eventParams;
    this.idMap = idMap;
    this.executor = executor;
    this.logger = logger;
    this.numOfThread = numOfThread > 1 ? numOfThread : 1;
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
private void doUserIndexExists(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap, final RequestHandlerChain chain) {
    final String index = params.param(
            TasteConstants.REQUEST_PARAM_USER_INDEX, params.param("index"));

    try {
        indexCreationLock.lock();
        final IndicesExistsResponse indicesExistsResponse = client.admin()
                .indices().prepareExists(index).execute().actionGet();
        if (indicesExistsResponse.isExists()) {
            doUserMappingCreation(params, listener, requestMap, paramMap,
                    chain);
        } else {
            doUserIndexCreation(params, listener, requestMap, paramMap,
                    chain, index);
        }
    } catch (final Exception e) {
        final List<Throwable> errorList = getErrorList(paramMap);
        if (errorList.size() >= maxRetryCount) {
            listener.onError(e);
        } else {
            sleep(e);
            errorList.add(e);
            fork(() -> execute(params, listener, requestMap, paramMap,
                    chain));
        }
    } finally {
        indexCreationLock.unlock();
    }
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
private void doUserUpdate(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap,
        final Map<String, Object> userMap, final String index,
        final String type, final String userIdField,
        final String timestampField, final Long userId,
        final OpType opType, final RequestHandlerChain chain) {
    userMap.put(userIdField, userId);
    userMap.put(timestampField, new Date());

    final OnResponseListener<IndexResponse> responseListener = response -> {
        paramMap.put(userIdField, userId);
        chain.execute(params, listener, requestMap, paramMap);
    };
    final OnFailureListener failureListener = t -> {
        if (t instanceof DocumentAlreadyExistsException
                || t instanceof EsRejectedExecutionException) {
            sleep(t);
            execute(params, listener, requestMap, paramMap, chain);
        } else {
            listener.onError(t);
        }
    };
    client.prepareIndex(index, type, userId.toString()).setSource(userMap)
            .setRefresh(true).setOpType(opType)
            .execute(on(responseListener, failureListener));
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
private void doItemMappingCreation(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap, final RequestHandlerChain chain) {
    final String index = params.param(
            TasteConstants.REQUEST_PARAM_ITEM_INDEX, params.param("index"));
    final String type = params.param(
            TasteConstants.REQUEST_PARAM_ITEM_TYPE,
            TasteConstants.ITEM_TYPE);
    final String itemIdField = params.param(
            TasteConstants.REQUEST_PARAM_ITEM_ID_FIELD,
            TasteConstants.ITEM_ID_FIELD);
    final String timestampField = params.param(
            TasteConstants.REQUEST_PARAM_TIMESTAMP_FIELD,
            TasteConstants.TIMESTAMP_FIELD);

    try (XContentBuilder jsonBuilder = XContentFactory.jsonBuilder()) {
        final ClusterHealthResponse healthResponse = client
                .admin()
                .cluster()
                .prepareHealth(index)
                .setWaitForYellowStatus()
                .setTimeout(
                        params.param("timeout",
                                DEFAULT_HEALTH_REQUEST_TIMEOUT)).execute()
                .actionGet();
        if (healthResponse.isTimedOut()) {
            listener.onError(new OperationFailedException(
                    "Failed to create index: " + index + "/" + type));
        }

        final XContentBuilder builder = jsonBuilder//
                .startObject()//
                .startObject(type)//
                .startObject("properties")//

                // @timestamp
                .startObject(timestampField)//
                .field("type", "date")//
                .field("format", "date_optional_time")//
                .endObject()//

                // item_id
                .startObject(itemIdField)//
                .field("type", "long")//
                .endObject()//

                // system_id
                .startObject("system_id")//
                .field("type", "string")//
                .field("index", "not_analyzed")//
                .endObject()//

                .endObject()//
                .endObject()//
                .endObject();

        final PutMappingResponse mappingResponse = client.admin().indices()
                .preparePutMapping(index).setType(type).setSource(builder)
                .execute().actionGet();
        if (mappingResponse.isAcknowledged()) {
            fork(() -> execute(params, listener, requestMap, paramMap,
                    chain));
        } else {
            listener.onError(new OperationFailedException(
                    "Failed to create mapping for " + index + "/" + type));
        }
    } catch (final Exception e) {
        listener.onError(e);
    }
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
private void doItemCreation(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap,
        final Map<String, Object> itemMap, final String index,
        final String type, final String itemIdField,
        final String timestampField, final RequestHandlerChain chain) {
    final OnResponseListener<SearchResponse> responseListener = response -> {
        validateRespose(response);

        Number currentId = null;
        final SearchHits hits = response.getHits();
        if (hits.getTotalHits() != 0) {
            final SearchHit[] searchHits = hits.getHits();
            final SearchHitField field = searchHits[0].getFields().get(
                    itemIdField);
            if (field != null) {
                currentId = field.getValue();
            }
        }
        final Long itemId;
        if (currentId == null) {
            itemId = Long.valueOf(1);
        } else {
            itemId = Long.valueOf(currentId.longValue() + 1);
        }
        doItemUpdate(params, listener, requestMap, paramMap, itemMap,
                index, type, itemIdField, timestampField, itemId,
                OpType.CREATE, chain);
    };
    final OnFailureListener failureListener = t -> {
        final List<Throwable> errorList = getErrorList(paramMap);
        if (errorList.size() >= maxRetryCount) {
            listener.onError(t);
        } else {
            sleep(t);
            errorList.add(t);
            doItemIndexExists(params, listener, requestMap, paramMap,
                    chain);
        }
    };
    client.prepareSearch(index).setTypes(type)
            .setQuery(QueryBuilders.matchAllQuery()).addField(itemIdField)
            .addSort(itemIdField, SortOrder.DESC).setSize(1)
            .execute(on(responseListener, failureListener));
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
private void doPreferenceMappingCreation(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap, final RequestHandlerChain chain) {
    final String index = params.param("index");
    final String type = params
            .param("type", TasteConstants.PREFERENCE_TYPE);
    final String userIdField = params.param(
            TasteConstants.REQUEST_PARAM_USER_ID_FIELD,
            TasteConstants.USER_ID_FIELD);
    final String itemIdField = params.param(
            TasteConstants.REQUEST_PARAM_ITEM_ID_FIELD,
            TasteConstants.ITEM_ID_FIELD);
    final String valueField = params.param(
            TasteConstants.REQUEST_PARAM_VALUE_FIELD,
            TasteConstants.VALUE_FIELD);
    final String timestampField = params.param(
            TasteConstants.REQUEST_PARAM_TIMESTAMP_FIELD,
            TasteConstants.TIMESTAMP_FIELD);

    try (XContentBuilder jsonBuilder = XContentFactory.jsonBuilder()) {
        final ClusterHealthResponse healthResponse = client
                .admin()
                .cluster()
                .prepareHealth(index)
                .setWaitForYellowStatus()
                .setTimeout(
                        params.param("timeout",
                                DEFAULT_HEALTH_REQUEST_TIMEOUT)).execute()
                .actionGet();
        if (healthResponse.isTimedOut()) {
            listener.onError(new OperationFailedException(
                    "Failed to create index: " + index + "/" + type));
        }

        final XContentBuilder builder = jsonBuilder//
                .startObject()//
                .startObject(type)//
                .startObject("properties")//

                // @timestamp
                .startObject(timestampField)//
                .field("type", "date")//
                .field("format", "date_optional_time")//
                .endObject()//

                // user_id
                .startObject(userIdField)//
                .field("type", "long")//
                .endObject()//

                // item_id
                .startObject(itemIdField)//
                .field("type", "long")//
                .endObject()//

                // value
                .startObject(valueField)//
                .field("type", "double")//
                .endObject()//

                .endObject()//
                .endObject()//
                .endObject();

        final PutMappingResponse mappingResponse = client.admin().indices()
                .preparePutMapping(index).setType(type).setSource(builder)
                .execute().actionGet();
        if (mappingResponse.isAcknowledged()) {
            fork(() -> execute(params, listener, requestMap, paramMap,
                    chain));
        } else {
            listener.onError(new OperationFailedException(
                    "Failed to create mapping for " + index + "/" + type));
        }
    } catch (final Exception e) {
        listener.onError(e);
    }
}
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
@Override
public abstract void execute(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap, RequestHandlerChain chain);
 

import org.elasticsearch.common.xcontent.ToXContent.Params; //导入依赖的package包/类
private void doUserMappingCreation(final Params params,
        final RequestHandler.OnErrorListener listener,
        final Map<String, Object> requestMap,
        final Map<String, Object> paramMap, final RequestHandlerChain chain) {
    final String index = params.param(
            TasteConstants.REQUEST_PARAM_USER_INDEX, params.param("index"));
    final String type = params.param(
            TasteConstants.REQUEST_PARAM_USER_TYPE,
            TasteConstants.USER_TYPE);
    final String userIdField = params.param(
            TasteConstants.REQUEST_PARAM_USER_ID_FIELD,
            TasteConstants.USER_ID_FIELD);
    final String timestampField = params.param(
            TasteConstants.REQUEST_PARAM_TIMESTAMP_FIELD,
            TasteConstants.TIMESTAMP_FIELD);

    try (XContentBuilder jsonBuilder = XContentFactory.jsonBuilder()) {
        final ClusterHealthResponse healthResponse = client
                .admin()
                .cluster()
                .prepareHealth(index)
                .setWaitForYellowStatus()
                .setTimeout(
                        params.param("timeout",
                                DEFAULT_HEALTH_REQUEST_TIMEOUT)).execute()
                .actionGet();
        if (healthResponse.isTimedOut()) {
            listener.onError(new OperationFailedException(
                    "Failed to create index: " + index + "/" + type));
        }

        final XContentBuilder builder = jsonBuilder//
                .startObject()//
                .startObject(type)//
                .startObject("properties")//

                // @timestamp
                .startObject(timestampField)//
                .field("type", "date")//
                .field("format", "date_optional_time")//
                .endObject()//

                // user_id
                .startObject(userIdField)//
                .field("type", "long")//
                .endObject()//

                // system_id
                .startObject("system_id")//
                .field("type", "string")//
                .field("index", "not_analyzed")//
                .endObject()//

                .endObject()//
                .endObject()//
                .endObject();

        final PutMappingResponse mappingResponse = client.admin().indices()
                .preparePutMapping(index).setType(type).setSource(builder)
                .execute().actionGet();
        if (mappingResponse.isAcknowledged()) {
            fork(() -> execute(params, listener, requestMap, paramMap,
                    chain));
        } else {
            listener.onError(new OperationFailedException(
                    "Failed to create mapping for " + index + "/" + type));
        }
    } catch (final Exception e) {
        listener.onError(e);
    }
}