Introduce IndexOperationsAdapter.

src/main/java/org/springframework/data/elasticsearch/core/IndexOperationsAdapter.java

@@ -0,0 +1,214 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core;
+
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.springframework.data.elasticsearch.core.document.Document;
+import org.springframework.data.elasticsearch.core.index.*;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Adapter for creating synchronous calls for a reactive {@link ReactiveIndexOperations}.
+ *
+ * @author Peter-Josef Meisch
+ * @since 5.2
+ */
+public interface IndexOperationsAdapter extends IndexOperations {
+	static IndexOperationsAdapter blocking(ReactiveIndexOperations reactiveIndexOperations) {
+
+		Assert.notNull(reactiveIndexOperations, "reactiveIndexOperations must not be null");
+
+		return new IndexOperationsAdapter() {
+			@Override
+			public boolean create() {
+				return Boolean.TRUE.equals(reactiveIndexOperations.create().block());
+			}
+
+			@Override
+			public boolean create(Map<String, Object> settings) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.create(settings).block());
+			}
+
+			@Override
+			public boolean create(Map<String, Object> settings, Document mapping) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.create(settings, mapping).block());
+			}
+
+			@Override
+			public boolean createWithMapping() {
+				return Boolean.TRUE.equals(reactiveIndexOperations.createWithMapping().block());
+			}
+
+			@Override
+			public boolean delete() {
+				return Boolean.TRUE.equals(reactiveIndexOperations.delete().block());
+			}
+
+			@Override
+			public boolean exists() {
+				return Boolean.TRUE.equals(reactiveIndexOperations.exists().block());
+			}
+
+			@Override
+			public void refresh() {
+				reactiveIndexOperations.refresh().block();
+			}
+
+			@Override
+			public Document createMapping() {
+				return Objects.requireNonNull(reactiveIndexOperations.createMapping().block());
+			}
+
+			@Override
+			public Document createMapping(Class<?> clazz) {
+				return Objects.requireNonNull(reactiveIndexOperations.createMapping(clazz).block());
+
+			}
+
+			@Override
+			public boolean putMapping(Document mapping) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.putMapping(Mono.just(mapping)).block());
+			}
+
+			@Override
+			public Map<String, Object> getMapping() {
+				return Objects.requireNonNull(reactiveIndexOperations.getMapping().block());
+			}
+
+			@Override
+			public Settings createSettings() {
+				return Objects.requireNonNull(reactiveIndexOperations.createSettings().block());
+			}
+
+			@Override
+			public Settings createSettings(Class<?> clazz) {
+				return Objects.requireNonNull(reactiveIndexOperations.createSettings(clazz).block());
+			}
+
+			@Override
+			public Settings getSettings() {
+				return Objects.requireNonNull(reactiveIndexOperations.getSettings().block());
+			}
+
+			@Override
+			public Settings getSettings(boolean includeDefaults) {
+				return Objects.requireNonNull(reactiveIndexOperations.getSettings(includeDefaults).block());
+			}
+
+			@Override
+			public boolean alias(AliasActions aliasActions) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.alias(aliasActions).block());
+			}
+
+			@Override
+			public Map<String, Set<AliasData>> getAliases(String... aliasNames) {
+				return Objects.requireNonNull(reactiveIndexOperations.getAliases(aliasNames).block());
+			}
+
+			@Override
+			public Map<String, Set<AliasData>> getAliasesForIndex(String... indexNames) {
+				return Objects.requireNonNull(reactiveIndexOperations.getAliasesForIndex(indexNames).block());
+			}
+
+			@Deprecated
+			@Override
+			public boolean putTemplate(PutTemplateRequest putTemplateRequest) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.putTemplate(putTemplateRequest).block());
+			}
+
+			@Override
+			public boolean putIndexTemplate(PutIndexTemplateRequest putIndexTemplateRequest) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.putIndexTemplate(putIndexTemplateRequest).block());
+			}
+
+			@Override
+			public boolean putComponentTemplate(PutComponentTemplateRequest putComponentTemplateRequest) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.putComponentTemplate(putComponentTemplateRequest).block());
+			}
+
+			@Override
+			public boolean existsComponentTemplate(ExistsComponentTemplateRequest existsComponentTemplateRequest) {
+				return Boolean.TRUE
+						.equals(reactiveIndexOperations.existsComponentTemplate(existsComponentTemplateRequest).block());
+			}
+
+			@Override
+			public List<TemplateResponse> getComponentTemplate(GetComponentTemplateRequest getComponentTemplateRequest) {
+				return Objects.requireNonNull(
+						reactiveIndexOperations.getComponentTemplate(getComponentTemplateRequest).collectList().block());
+			}
+
+			@Override
+			public boolean deleteComponentTemplate(DeleteComponentTemplateRequest deleteComponentTemplateRequest) {
+				return Boolean.TRUE
+						.equals(reactiveIndexOperations.deleteComponentTemplate(deleteComponentTemplateRequest).block());
+			}
+
+			@Deprecated
+			@Nullable
+			@Override
+			public TemplateData getTemplate(GetTemplateRequest getTemplateRequest) {
+				return Objects.requireNonNull(reactiveIndexOperations).getTemplate(getTemplateRequest).block();
+			}
+
+			@Deprecated
+			@Override
+			public boolean existsTemplate(ExistsTemplateRequest existsTemplateRequest) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.existsTemplate(existsTemplateRequest).block());
+			}
+
+			@Override
+			public boolean existsIndexTemplate(ExistsIndexTemplateRequest existsTemplateRequest) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.existsIndexTemplate(existsTemplateRequest).block());
+			}
+
+			@Override
+			public List<TemplateResponse> getIndexTemplate(GetIndexTemplateRequest getIndexTemplateRequest) {
+				return Objects
+						.requireNonNull(reactiveIndexOperations.getIndexTemplate(getIndexTemplateRequest).collectList().block());
+			}
+
+			@Override
+			public boolean deleteIndexTemplate(DeleteIndexTemplateRequest deleteIndexTemplateRequest) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.deleteIndexTemplate(deleteIndexTemplateRequest).block());
+			}
+
+			@Deprecated
+			@Override
+			public boolean deleteTemplate(DeleteTemplateRequest deleteTemplateRequest) {
+				return Boolean.TRUE.equals(reactiveIndexOperations.deleteTemplate(deleteTemplateRequest).block());
+			}
+
+			@Override
+			public List<IndexInformation> getInformation(IndexCoordinates index) {
+				return Objects.requireNonNull(reactiveIndexOperations.getInformation(index).collectList().block());
+			}
+
+			@Override
+			public IndexCoordinates getIndexCoordinates() {
+				return reactiveIndexOperations.getIndexCoordinates();
+			}
+		};
+	}
+}

src/main/java/org/springframework/data/elasticsearch/repository/support/SimpleReactiveElasticsearchRepository.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.elasticsearch.repository.support;
 
+import static org.springframework.data.elasticsearch.core.IndexOperationsAdapter.*;
+
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -62,12 +64,14 @@ public SimpleReactiveElasticsearchRepository(ElasticsearchEntityInformation<T, I
 
 	private void createIndexAndMappingIfNeeded() {
 
+		var blockingIndexOperations = blocking(indexOperations);
+
 		if (shouldCreateIndexAndMapping()) {
-			indexOperations.exists() //
-					.flatMap(exists -> exists ? Mono.empty() : indexOperations.createWithMapping()) //
-					.block();
-		} else if(shouldAlwaysWriteMapping()) {
-			indexOperations.putMapping().block();
+			if (!blockingIndexOperations.exists()) {
+				blockingIndexOperations.createWithMapping();
+			}
+		} else if (shouldAlwaysWriteMapping()) {
+			blockingIndexOperations.putMapping();
 		}
 	}
 

src/test/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchIntegrationTests.java

@@ -18,6 +18,7 @@
 import static java.util.Collections.*;
 import static org.assertj.core.api.Assertions.*;
 import static org.springframework.data.elasticsearch.annotations.FieldType.*;
+import static org.springframework.data.elasticsearch.core.IndexOperationsAdapter.*;
 import static org.springframework.data.elasticsearch.core.query.StringQuery.*;
 
 import reactor.core.publisher.Flux;
@@ -105,14 +106,14 @@ public abstract class ReactiveElasticsearchIntegrationTests {
 	public void beforeEach() {
 
 		indexNameProvider.increment();
-		operations.indexOps(SampleEntity.class).createWithMapping().block();
-		operations.indexOps(IndexedIndexNameEntity.class).createWithMapping().block();
+		blocking(operations.indexOps(SampleEntity.class)).createWithMapping();
+		blocking(operations.indexOps(IndexedIndexNameEntity.class)).createWithMapping();
 	}
 
 	@Test
 	@Order(java.lang.Integer.MAX_VALUE)
 	void cleanup() {
-		operations.indexOps(IndexCoordinates.of(indexNameProvider.getPrefix() + "*")).delete().block();
+		blocking(operations.indexOps(IndexCoordinates.of(indexNameProvider.getPrefix() + '*'))).delete();
 	}
 	// endregion
 
@@ -625,7 +626,8 @@ public void shouldDeleteAcrossIndex() {
 				.as(StepVerifier::create)//
 				.verifyComplete();
 
-		operations.indexOps(thisIndex).refresh().then(operations.indexOps(thatIndex).refresh()).block();
+		blocking(operations.indexOps(thisIndex)).refresh();
+		blocking(operations.indexOps(thatIndex)).refresh();
 
 		Query query = getBuilderWithTermQuery("message", "test").build();
 
@@ -651,7 +653,8 @@ public void shouldDeleteAcrossIndexWhenNoMatchingDataPresent() {
 				.as(StepVerifier::create)//
 				.verifyComplete();
 
-		operations.indexOps(thisIndex).refresh().then(operations.indexOps(thatIndex).refresh()).block();
+		blocking(operations.indexOps(thisIndex)).refresh();
+		blocking(operations.indexOps(thatIndex)).refresh();
 
 		Query query = getBuilderWithTermQuery("message", "negative").build();
 
@@ -876,7 +879,7 @@ void searchShouldReturnSeqNoPrimaryTerm() {
 		original.setMessage("It's fine");
 		OptimisticEntity saved = operations.save(original).block();
 
-		operations.indexOps(OptimisticEntity.class).refresh().block();
+		blocking(operations.indexOps(OptimisticEntity.class)).refresh();
 
 		operations
 				.search(searchQueryForOne(saved.getId()), OptimisticEntity.class,
@@ -1081,14 +1084,16 @@ void shouldReturnInformationListOfAllIndices() {
 		String indexName = indexNameProvider.indexName();
 		String aliasName = indexName + "-alias";
 		ReactiveIndexOperations indexOps = operations.indexOps(EntityWithSettingsAndMappingsReactive.class);
+		var blockingIndexOps = blocking(indexOps);
 
 		// beforeEach uses SampleEntity, so recreate the index here
-		indexOps.delete().then(indexOps.createWithMapping()).block();
+		blockingIndexOps.delete();
+		blockingIndexOps.createWithMapping();
 
 		AliasActionParameters parameters = AliasActionParameters.builder().withAliases(aliasName).withIndices(indexName)
 				.withIsHidden(false).withIsWriteIndex(false).withRouting("indexrouting").withSearchRouting("searchrouting")
 				.build();
-		indexOps.alias(new AliasActions(new AliasAction.Add(parameters))).block();
+		blockingIndexOps.alias(new AliasActions(new AliasAction.Add(parameters)));
 
 		indexOps.getInformation().as(StepVerifier::create).consumeNextWith(indexInformation -> {
 			assertThat(indexInformation.getName()).isEqualTo(indexName);

src/test/java/org/springframework/data/elasticsearch/core/ReactivePointInTimeIntegrationTests.java

@@ -16,6 +16,7 @@
 package org.springframework.data.elasticsearch.core;
 
 import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.elasticsearch.core.IndexOperationsAdapter.*;
 
 import java.time.Duration;
 import java.util.List;
@@ -49,19 +50,17 @@ public abstract class ReactivePointInTimeIntegrationTests {
 
 	@Autowired ReactiveElasticsearchOperations operations;
 	@Autowired IndexNameProvider indexNameProvider;
-	@Nullable ReactiveIndexOperations indexOperations;
 
 	@BeforeEach
 	void setUp() {
 		indexNameProvider.increment();
-		indexOperations = operations.indexOps(SampleEntity.class);
-		indexOperations.createWithMapping().block();
+		blocking(operations.indexOps(SampleEntity.class)).createWithMapping();
 	}
 
 	@Test
 	@Order(Integer.MAX_VALUE)
 	void cleanup() {
-		operations.indexOps(IndexCoordinates.of(indexNameProvider.getPrefix() + '*')).delete().block();
+		blocking(operations.indexOps(IndexCoordinates.of(indexNameProvider.getPrefix() + '*'))).delete();
 	}
 
 	@Test // #1684

src/test/java/org/springframework/data/elasticsearch/core/ReactiveReindexIntegrationTests.java

@@ -16,6 +16,7 @@
 package org.springframework.data.elasticsearch.core;
 
 import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.elasticsearch.core.IndexOperationsAdapter.*;
 import static org.springframework.data.elasticsearch.utils.IdGenerator.*;
 
 import reactor.test.StepVerifier;
@@ -54,13 +55,13 @@ public abstract class ReactiveReindexIntegrationTests {
 	public void beforeEach() {
 
 		indexNameProvider.increment();
-		operations.indexOps(Entity.class).createWithMapping().block();
+		blocking(operations.indexOps(Entity.class)).createWithMapping();
 	}
 
 	@Test
 	@Order(java.lang.Integer.MAX_VALUE)
 	void cleanup() {
-		operations.indexOps(IndexCoordinates.of(indexNameProvider.getPrefix() + "*")).delete().block();
+		blocking(operations.indexOps(IndexCoordinates.of(indexNameProvider.getPrefix() + '*'))).delete();
 	}
 
 	@Test