Add count methods to ELC's ReactiveElasticsearchClient.

Implemented count methods using CountRequest directly with ELC's ReactiveElasticsearchClient, eliminating the need to initialize any entity or repository.
Refactored the existing doCount method in ReactiveElasticsearchTemplate to utilize the newly added count method.

Closes spring-projects#2749

Author: maryantocinn

Diff for: 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,26 @@ public Mono<DeleteByQueryResponse> deleteByQuery(
 		return deleteByQuery(fn.apply(new DeleteByQueryRequest.Builder()).build());
 	}
 
+	/**
+	 * @since 5.4
+	 */
+	public Mono<CountResponse> count(CountRequest request) {
+
+		Assert.notNull(request, "request must not be null");
+
+		return Mono.fromFuture(transport.performRequestAsync(request, CountRequest._ENDPOINT, transportOptions));
+	}
+
+	/**
+	 * @since 5.4
+	 */
+	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
 

Diff for: 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) {

Diff for: 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

Diff for: src/test/java/org/springframework/data/elasticsearch/core/ReactiveCountELCIntegrationTests.java

@@ -0,0 +1,25 @@
+package org.springframework.data.elasticsearch.core;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.elasticsearch.junit.jupiter.ReactiveElasticsearchTemplateConfiguration;
+import org.springframework.data.elasticsearch.utils.IndexNameProvider;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * @author maryantocinn
+ * @since 5.4
+ */
+@ContextConfiguration(classes = ReactiveCountELCIntegrationTests.Config.class)
+public class ReactiveCountELCIntegrationTests extends ReactiveCountIntegrationTests {
+
+	@Configuration
+	@Import({ ReactiveElasticsearchTemplateConfiguration.class })
+	static class Config {
+		@Bean
+		IndexNameProvider indexNameProvider() {
+			return new IndexNameProvider("reactive-count");
+		}
+	}
+}

Diff for: src/test/java/org/springframework/data/elasticsearch/core/ReactiveCountIntegrationTests.java

@@ -0,0 +1,101 @@
+package org.springframework.data.elasticsearch.core;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.annotations.Field;
+import org.springframework.data.elasticsearch.annotations.FieldType;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
+import org.springframework.data.elasticsearch.utils.IndexNameProvider;
+import org.springframework.lang.Nullable;
+import reactor.test.StepVerifier;
+
+import static org.springframework.data.elasticsearch.core.IndexOperationsAdapter.*;
+import static org.springframework.data.elasticsearch.utils.IdGenerator.*;
+
+/**
+ * Integration tests for the count API.
+ *
+ * @author maryantocinn
+ * @since 5.4
+ */
+@SpringIntegrationTest
+public abstract class ReactiveCountIntegrationTests {
+
+	@Autowired private ReactiveElasticsearchOperations operations;
+	@Autowired private IndexNameProvider indexNameProvider;
+
+	@BeforeEach
+	public void beforeEach() {
+
+		indexNameProvider.increment();
+		blocking(operations.indexOps(ReactiveReindexIntegrationTests.Entity.class)).createWithMapping();
+	}
+
+	@Test
+	@Order(java.lang.Integer.MAX_VALUE)
+	void cleanup() {
+		blocking(operations.indexOps(IndexCoordinates.of(indexNameProvider.getPrefix() + '*'))).delete();
+	}
+
+	@Test // GH-2749
+	void shouldCount() {
+		String indexName = indexNameProvider.indexName();
+
+		Entity entity = new Entity();
+		entity.setId(nextIdAsString());
+		entity.setMessage("abc");
+
+		operations.save(entity) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		Entity entity2 = new Entity();
+		entity2.setId(nextIdAsString());
+		entity2.setMessage("def");
+
+		operations.save(entity2) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		operations.count(operations.matchAllQuery(), Entity.class, IndexCoordinates.of(indexName)) //
+				.as(StepVerifier::create) //
+				.expectNext(2L) //
+				.verifyComplete();
+	}
+
+
+	@Document(indexName = "#{@indexNameProvider.indexName()}")
+	static class Entity {
+		@Nullable
+		@Id
+		private String id;
+		@Nullable
+		@Field(type = FieldType.Text) private String message;
+
+		@Nullable
+		public String getId() {
+			return id;
+		}
+
+		public void setId(@Nullable String id) {
+			this.id = id;
+		}
+
+		@Nullable
+		public String getMessage() {
+			return message;
+		}
+
+		public void setMessage(@Nullable String message) {
+			this.message = message;
+		}
+	}
+
+}