import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Override
public void doRequest(final RestRequest request, final RestChannel channel, final Client client) {
    String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
    CountRequest countRequest = new CountRequest(indices);
    String source = request.param("source");
    if (source != null) {
        countRequest.source(source);
    } else {
        QuerySourceBuilder querySourceBuilder = RestActions.parseQuerySource(request);
        if (querySourceBuilder != null) {
            countRequest.source(querySourceBuilder);
        }
    }

    client.search(countRequest.toSearchRequest(), new RestResponseListener<SearchResponse>(channel) {
        @Override
        public RestResponse buildResponse(SearchResponse countResponse) throws Exception {
            return RestTable.buildResponse(buildTable(request, countResponse), channel);
        }
    });
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Test
public void should_count() throws IOException, ExecutionException, InterruptedException {
    BytesReference source = source().bytes();
    Map<String, Object> expected = SourceLookup.sourceAsMap(source);
    index(THE_INDEX, THE_TYPE, THE_ID, expected);

    refresh();

    CountRequest countRequest = new CountRequest(THE_INDEX).types(THE_TYPE).source(new QuerySourceBuilder().setQuery(matchAllQuery()));
    CountResponse countResponse = httpClient.count(countRequest).get();

    Shards shards = countResponse.getShards();
    Assertions.assertThat(shards.getTotal()).isEqualTo(getNumShards(THE_INDEX).numPrimaries);
    Assertions.assertThat(shards.getSuccessful()).isEqualTo(getNumShards(THE_INDEX).numPrimaries);
    Assertions.assertThat(shards.getFailed()).isEqualTo(0);
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
protected long getCount(final List<String> indices, final String type) {
    logger.debug("getCount() for index {} and type", indices, type);
    
    esSetup.client().admin().indices().refresh(new RefreshRequest()).actionGet();

    long count = 0;
    
    for (Iterator<String> iterator = indices.iterator(); iterator.hasNext();) {
        String index = (String) iterator.next();
         long lcount = esSetup.client().count(new CountRequest(index).types(type)).actionGet().getCount();
         logger.debug("Count for index {} (type {}) is {}", index, type, lcount);
         count += lcount;
    }

    return count;
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Override
public ActionFuture<CountResponse> count(final CountRequest request) {
    AdapterActionFuture<CountResponse, SearchResponse> actionFuture = new AdapterActionFuture<CountResponse, SearchResponse>() {
        @Override
        protected CountResponse convert(SearchResponse listenerResponse) {
            return new CountResponse(listenerResponse);
        }
    };
    deprecationLogger.deprecated("the count api is deprecated and will be removed from the java api in the next major version");
    execute(SearchAction.INSTANCE, request.toSearchRequest(), actionFuture);
    return actionFuture;
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) {
    CountRequest countRequest = new CountRequest(Strings.splitStringByCommaToArray(request.param("index")));
    countRequest.indicesOptions(IndicesOptions.fromRequest(request, countRequest.indicesOptions()));
    if (RestActions.hasBodyContent(request)) {
        countRequest.source(RestActions.getRestContent(request));
    } else {
        QuerySourceBuilder querySourceBuilder = RestActions.parseQuerySource(request);
        if (querySourceBuilder != null) {
            countRequest.source(querySourceBuilder);
        }
    }
    countRequest.routing(request.param("routing"));
    countRequest.minScore(request.paramAsFloat("min_score", DEFAULT_MIN_SCORE));
    countRequest.types(Strings.splitStringByCommaToArray(request.param("type")));
    countRequest.preference(request.param("preference"));

    final int terminateAfter = request.paramAsInt("terminate_after", DEFAULT_TERMINATE_AFTER);
    if (terminateAfter < 0) {
        throw new IllegalArgumentException("terminateAfter must be > 0");
    } else if (terminateAfter > 0) {
        countRequest.terminateAfter(terminateAfter);
    }
    client.search(countRequest.toSearchRequest(), new RestBuilderListener<SearchResponse>(channel) {
        @Override
        public RestResponse buildResponse(SearchResponse response, XContentBuilder builder) throws Exception {
            builder.startObject();
            if (terminateAfter != DEFAULT_TERMINATE_AFTER) {
                builder.field("terminated_early", response.isTerminatedEarly());
            }
            builder.field("count", response.getHits().totalHits());
            buildBroadcastShardsHeader(builder, request, response.getTotalShards(), response.getSuccessfulShards(),
                    response.getFailedShards(), response.getShardFailures());

            builder.endObject();
            return new BytesRestResponse(response.status(), builder);
        }
    });
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Test
public void should_fail_on_invalid_query() throws IOException, ExecutionException, InterruptedException {
    CountRequest countRequest = new CountRequest(THE_INDEX).types(THE_TYPE).source("{invalid query}");
    try {
        httpClient.count(countRequest).get();
        fail();
    } catch (ExecutionException e) {
        Assertions.assertThat(e).hasCauseInstanceOf(ElasticsearchHttpException.class);
        Assertions.assertThat(e.getMessage()).contains("status code 400");
        Assertions.assertThat(e.getMessage()).contains("Failed to parse");
        Assertions.assertThat(e.getMessage()).contains("QueryParsingException");
    }
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Test
public void testDefaultArgument() throws IOException {
    String index = "topic";

    createDefaultESSink(index);

    refresh();
    CountResponse countResponse = client().count(new CountRequest(index)).actionGet();
    assertEquals(countResponse.getCount(), 100);
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
public long getDocumentCount() throws Exception {
    try (Client client = getClient()) {
        client.admin().indices().refresh(new RefreshRequest(INDEX_NAME)).actionGet();

        ActionFuture<CountResponse> response = client.count(new CountRequest(INDEX_NAME).types(DOCUMENT_TYPE));
        CountResponse countResponse = response.get();
        return countResponse.getCount();
    }
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Override
protected XContentBuilder toXContent(CountRequest request, CountResponse response, XContentBuilder builder) throws IOException {
    builder.startObject();
    builder.field(Fields.COUNT, response.getCount());
    buildBroadcastShardsHeader(builder, response);
    builder.endObject();
    return builder;
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
private void validateDocCount(int expectedDocs) throws InterruptedException, ExecutionException {
	Client client = server.getClient();
	client.admin().indices().flush(new FlushRequest()).get();
	long docs = client.count(new CountRequest()).get().getCount();
	assertEquals(docs, expectedDocs);
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
public void count(CountRequest request, ActionListener<CountResponse> listener) {
    countActionHandler.execute(request, listener);
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Test
public void testIndexInfoBuilder() throws IOException {
    ObjectMapper jsonMapper = new DefaultObjectMapper();
    Properties props = new Properties();
    props.setProperty("dateFormat", "YYYYMMdd");
    ElasticSearchSink sink = new ElasticSearchSink(
        "testIndexInfoBuilder",
        null,
        1,
        1000,
        Lists.newArrayList("localhost:" + getPort()),
        new DefaultIndexInfoBuilder(
            null,
            null,
            new TimestampField("ts", null),
            new IndexSuffixFormatter("date", props),
            null,
            jsonMapper),
        0,0,0,0,0,
        null,
        false,
        jsonMapper,
        null
    );
    sink.open();

    DateTime dt = new DateTime("2014-10-12T12:12:12.000Z");

    Map<String, Object> msg = new ImmutableMap.Builder<String, Object>()
        .put("f1", "v1")
        .put("f2", "v2")
        .put("f3", "v3")
        .put("ts", dt.getMillis())
        .build();

    String routingKey = "topic";
    String index = "topic20141012";
    for (int i = 0; i < 100; ++i) {
        sink.writeTo(new DefaultMessageContainer(new Message(routingKey, jsonMapper.writeValueAsBytes(msg)), jsonMapper));
    }
    sink.close();

    refresh();
    CountResponse countResponse = client().count(new CountRequest(index)).actionGet();
    assertEquals(countResponse.getCount(), 100);
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Test
public void testRecover() throws Exception {
    ObjectMapper jsonMapper = new DefaultObjectMapper();
    ElasticSearchSink sink = new ElasticSearchSink(
        "default",
        null,
        10,
        1000,
        Lists.newArrayList("localhost:" + getPort()),
        null,
        0,0,0,0,
        0,
        null,
        false,
        jsonMapper,
        null
    );
    sink.open();

    DateTime dt = new DateTime("2014-10-12T12:12:12.000Z");

    Map<String, Object> msg = new ImmutableMap.Builder<String, Object>()
        .put("f1", "v1")
        .put("f2", "v2")
        .put("f3", "v3")
        .put("ts", dt.getMillis())
        .build();
    String routingKey = "topicrecover";
    String index = "topicrecover";
    List<Message> msgList = new ArrayList<>();
    int msgCount = 100;
    for (int i = 0; i < msgCount; ++i) {
        msgList.add(new Message(routingKey, jsonMapper.writeValueAsBytes(msg)));
    }

    for (Message m : msgList) {
        sink.recover(m);
    }

    refresh();
    CountResponse countResponse = client().count(new CountRequest(index)).actionGet();
    assertEquals(countResponse.getCount(), 100);
}
 

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

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
@Override
protected ActionFuture<CountResponse> doExecute(CountRequest request) {
    return client.count(request);
}
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
/**
 * A count of all the documents matching a specific query.
 *
 * @param request The count request
 * @return The result future
 * @see Requests#countRequest(String...)
 * @deprecated use {@link #search(SearchRequest)} instead and set size to 0
 */
@Deprecated
ActionFuture<CountResponse> count(CountRequest request);
 

import org.elasticsearch.action.count.CountRequest; //导入依赖的package包/类
/**
 * Creates a count request which counts the hits matched against a query. Note, the query itself must be set
 * either using the JSON source of the query, or using a {@link org.elasticsearch.index.query.QueryBuilder} (using {@link org.elasticsearch.index.query.QueryBuilders}).
 *
 * @param indices The indices to count matched documents against a query. Use <tt>null</tt> or <tt>_all</tt> to execute against all indices
 * @return The count request
 * @see org.elasticsearch.client.Client#count(org.elasticsearch.action.count.CountRequest)
 * @deprecated use {@link #searchRequest(String...)} instead and set size to 0
 */
@Deprecated
public static CountRequest countRequest(String... indices) {
    return new CountRequest(indices);
}