Add count methods in ELC ReactiveElasticsearchClient.

Add count methods using CountRequest with just ELC's ReactiveElasticsearchClient, without having to initialize any entity or repository. Also, refactored existing doCount method from ELC's ReactiveElasticsearchTemplate to use the newly created count method.

Closes spring-projects#2749

Author: maryantocinn

src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchClient.java

@@ -36,6 +36,7 @@
  * Reactive version of {@link co.elastic.clients.elasticsearch.ElasticsearchClient}.
  *
  * @author Peter-Josef Meisch
+ * @author maryantocinn
  * @since 4.4
  */
 public class ReactiveElasticsearchClient extends ApiClient<ElasticsearchTransport, ReactiveElasticsearchClient>
@@ -227,6 +228,20 @@ public Mono<DeleteByQueryResponse> deleteByQuery(
 		return deleteByQuery(fn.apply(new DeleteByQueryRequest.Builder()).build());
 	}
 
+	public Mono<CountResponse> count(CountRequest request) {
+
+		Assert.notNull(request, "request must not be null");
+		
+		return Mono.fromFuture(transport.performRequestAsync(request, CountRequest._ENDPOINT, transportOptions));
+	}
+
+	public Mono<CountResponse> count(Function<CountRequest.Builder, ObjectBuilder<CountRequest>> fn) {
+
+		Assert.notNull(fn, "fn must not be null");
+
+		return count(fn.apply(new CountRequest.Builder()).build());
+	}
+
 	// endregion
 	// region search
 

src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchTemplate.java

@@ -75,6 +75,7 @@
  * @author Peter-Josef Meisch
  * @author Illia Ulianov
  * @author Junghoon Ban
+ * @author maryantocinn
  * @since 4.4
  */
 public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearchTemplate {
@@ -483,11 +484,10 @@ protected Mono<Long> doCount(Query query, Class<?> entityType, IndexCoordinates
 		Assert.notNull(query, "query must not be null");
 		Assert.notNull(index, "index must not be null");
 
-		SearchRequest searchRequest = requestConverter.searchRequest(query, routingResolver.getRouting(), entityType, index,
-				true);
+		CountRequest request = requestConverter.countRequest(query, entityType, index);
 
-		return Mono.from(execute(client -> client.search(searchRequest, EntityAsMap.class)))
-				.map(searchResponse -> searchResponse.hits().total() != null ? searchResponse.hits().total().value() : 0L);
+		return Mono.from(execute(client -> client.count(request))) //
+				.map(CountResponse::count);
 	}
 
 	private Flux<SearchDocument> doFindBounded(Query query, Class<?> clazz, IndexCoordinates index) {

src/main/java/org/springframework/data/elasticsearch/client/elc/RequestConverter.java

@@ -111,6 +111,7 @@
  * @author cdalxndr
  * @author scoobyzhang
  * @author Haibo Liu
+ * @author maryantocinn
  * @since 4.4
  */
 class RequestConverter extends AbstractQueryProcessor {
@@ -1188,6 +1189,18 @@ public UpdateByQueryRequest documentUpdateByQueryRequest(UpdateQuery updateQuery
 		});
 	}
 
+	public <T> CountRequest countRequest(Query query, @Nullable Class<T> clazz, IndexCoordinates indexCoordinates) {
+
+		Assert.notNull(query, "query must not be null");
+		Assert.notNull(indexCoordinates, "index must not be null");
+
+		elasticsearchConverter.updateQuery(query, clazz);
+
+		return CountRequest.of(b -> b //
+				.query(getQuery(query, clazz)) //
+				.index(Arrays.asList(indexCoordinates.getIndexNames())));
+	}
+
 	// endregion
 
 	// region search