DATAES-738 - Add entity related save methods to DocumentOperations.

Original PR: #389

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

@@ -1,10 +1,13 @@
 package org.springframework.data.elasticsearch.core;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkResponse;
@@ -27,10 +30,13 @@ import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersiste
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext;
 import org.springframework.data.elasticsearch.core.query.DeleteQuery;
+import org.springframework.data.elasticsearch.core.query.IndexQuery;
+import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
 import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery;
 import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
 import org.springframework.data.elasticsearch.core.query.Query;
 import org.springframework.data.util.CloseableIterator;
+import org.springframework.data.util.Streamable;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -83,6 +89,68 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 	// endregion
 
 	// region DocumentOperations
+
+	@Override
+	public <T> T save(T entity) {
+
+		Assert.notNull(entity, "entity must not be null");
+
+		return save(entity, getIndexCoordinatesFor(entity.getClass()));
+	}
+
+	@Override
+	public <T> T save(T entity, IndexCoordinates index) {
+
+		Assert.notNull(entity, "entity must not be null");
+		Assert.notNull(index, "index must not be null");
+
+		index(getIndexQuery(entity), index);
+		return entity;
+	}
+
+	@Override
+	public <T> Iterable<T> save(Iterable<T> entities) {
+
+		Assert.notNull(entities, "entities must not be null");
+
+		Iterator<T> iterator = entities.iterator();
+		if (iterator.hasNext()) {
+			return save(entities, getIndexCoordinatesFor(iterator.next().getClass()));
+		}
+
+		return entities;
+	}
+
+	@Override
+	public <T> Iterable<T> save(Iterable<T> entities, IndexCoordinates index) {
+
+		Assert.notNull(entities, "entities must not be null");
+		Assert.notNull(index, "index must not be null");
+
+		List<IndexQuery> indexQueries = Streamable.of(entities).stream().map(this::getIndexQuery)
+				.collect(Collectors.toList());
+
+		if (!indexQueries.isEmpty()) {
+			List<String> ids = bulkIndex(indexQueries, index);
+			Iterator<String> idIterator = ids.iterator();
+			entities.forEach(entity -> {
+				setPersistentEntityId(entity, idIterator.next());
+			});
+		}
+
+		return entities;
+	}
+
+	@Override
+	public <T> Iterable<T> save(T... entities) {
+		return save(Arrays.asList(entities));
+	}
+
+	@Override
+	public <T> Iterable<T> save(IndexCoordinates index, T... entities) {
+		return save(Arrays.asList(entities), index);
+	}
+
 	@Override
 	public void delete(Query query, Class<?> clazz, IndexCoordinates index) {
 
@@ -197,7 +265,11 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 		return getRequiredPersistentEntity(clazz).getIndexCoordinates();
 	}
 
-	protected void checkForBulkOperationFailure(BulkResponse bulkResponse) {
+	/**
+	 * @param bulkResponse
+	 * @return the list of the item id's
+	 */
+	protected List<String> checkForBulkOperationFailure(BulkResponse bulkResponse) {
 
 		if (bulkResponse.hasFailures()) {
 			Map<String, String> failedDocuments = new HashMap<>();
@@ -211,6 +283,8 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 							+ failedDocuments + ']',
 					failedDocuments);
 		}
+
+		return Stream.of(bulkResponse.getItems()).map(BulkItemResponse::getId).collect(Collectors.toList());
 	}
 
 	protected void setPersistentEntityId(Object entity, String id) {
@@ -227,5 +301,39 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper
 	ElasticsearchPersistentEntity<?> getRequiredPersistentEntity(Class<?> clazz) {
 		return elasticsearchConverter.getMappingContext().getRequiredPersistentEntity(clazz);
 	}
+
+	@Nullable
+	private String getEntityId(Object entity) {
+		ElasticsearchPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
+		ElasticsearchPersistentProperty idProperty = persistentEntity.getIdProperty();
+
+		if (idProperty != null) {
+			return stringIdRepresentation(persistentEntity.getPropertyAccessor(entity).getProperty(idProperty));
+		}
+
+		return null;
+	}
+
+	@Nullable
+	private Long getEntityVersion(Object entity) {
+		ElasticsearchPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
+		ElasticsearchPersistentProperty versionProperty = persistentEntity.getVersionProperty();
+
+		if (versionProperty != null) {
+			Object version = persistentEntity.getPropertyAccessor(entity).getProperty(versionProperty);
+
+			if (version != null && Long.class.isAssignableFrom(version.getClass())) {
+				return ((Long) version);
+			}
+		}
+
+		return null;
+	}
+
+	private <T> IndexQuery getIndexQuery(T entity) {
+		return new IndexQueryBuilder().withObject(entity).withId(getEntityId(entity)).withVersion(getEntityVersion(entity))
+				.build();
+	}
+
 	// endregion
 }

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

@@ -25,6 +25,7 @@ import org.springframework.data.elasticsearch.core.query.GetQuery;
 import org.springframework.data.elasticsearch.core.query.IndexQuery;
 import org.springframework.data.elasticsearch.core.query.Query;
 import org.springframework.data.elasticsearch.core.query.UpdateQuery;
+import org.springframework.lang.Nullable;
 
 /**
  * The operations for the
@@ -35,6 +36,63 @@ import org.springframework.data.elasticsearch.core.query.UpdateQuery;
  */
 public interface DocumentOperations {
 
+	/**
+	 * Saves an entity to the index specified in the entity's Document annotation
+	 * 
+	 * @param entity the entity to save, must not be {@literal null}
+	 * @param <T> the entity type
+	 * @return the saved entity
+	 */
+	<T> T save(T entity);
+
+	/**
+	 * Saves an entity to the index specified in the entity's Document annotation
+	 * 
+	 * @param entity the entity to save, must not be {@literal null}
+	 * @param index the index to save the entity in, must not be {@literal null}
+	 * @param <T> the entity type
+	 * @return the saved entity
+	 */
+	<T> T save(T entity, IndexCoordinates index);
+
+	/**
+	 * saves the given entities to the index retrieved from the entities' Document annotation
+	 * 
+	 * @param entities must not be {@literal null}
+	 * @param <T> the entity type
+	 * @return the saved entites
+	 */
+	<T> Iterable<T> save(Iterable<T> entities);
+
+	/**
+	 * saves the given entities to the given index
+	 *
+	 * @param entities must not be {@literal null}
+	 * @param index the idnex to save the entities in, must not be {@literal null}
+	 * @param <T> the entity type
+	 * @return the saved entites
+	 */
+	<T> Iterable<T> save(Iterable<T> entities, IndexCoordinates index);
+
+	/**
+	 * saves the given entities to the index retrieved from the entities' Document annotation
+	 *
+	 * @param entities must not be {@literal null}
+	 * @param <T> the entity type
+	 * @return the saved entites as Iterable
+	 */
+	<T> Iterable<T> save(T... entities);
+
+	/**
+	 * saves the given entities to the given index.
+	 *
+	 * @param index the idnex to save the entities in, must not be {@literal null}
+	 * @param entities must not be {@literal null}
+	 * @param <T> the entity type
+	 * @return the saved entites as Iterable
+	 */
+	<T> Iterable<T> save(IndexCoordinates index, T... entities);
+
 	/**
 	 * Index an object. Will do save or update.
 	 *
@@ -52,6 +110,7 @@ public interface DocumentOperations {
 	 * @param index the index from which the object is read.
 	 * @return the found object
 	 */
+	@Nullable
 	<T> T get(GetQuery query, Class<T> clazz, IndexCoordinates index);
 
 	/**
@@ -68,9 +127,10 @@ public interface DocumentOperations {
 	 * Bulk index all objects. Will do save or update.
 	 *
 	 * @param queries the queries to execute in bulk
+	 * @return the ids of the indexed objects
 	 */
-	default void bulkIndex(List<IndexQuery> queries, IndexCoordinates index) {
-		bulkIndex(queries, BulkOptions.defaultOptions(), index);
+	default List<String> bulkIndex(List<IndexQuery> queries, IndexCoordinates index) {
+		return bulkIndex(queries, BulkOptions.defaultOptions(), index);
 	}
 
 	/**
@@ -78,8 +138,9 @@ public interface DocumentOperations {
 	 * 
 	 * @param queries the queries to execute in bulk
 	 * @param bulkOptions options to be added to the bulk request
+	 * @return the ids of the indexed objects
 	 */
-	void bulkIndex(List<IndexQuery> queries, BulkOptions bulkOptions, IndexCoordinates index);
+	List<String> bulkIndex(List<IndexQuery> queries, BulkOptions bulkOptions, IndexCoordinates index);
 
 	/**
 	 * Bulk update all objects. Will do update.

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

@@ -17,11 +17,13 @@ package org.springframework.data.elasticsearch.core;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 import org.elasticsearch.cluster.metadata.AliasMetaData;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.AliasQuery;
+import org.springframework.lang.Nullable;
 
 /**
  * ElasticsearchOperations. Since 4.0 this interface only contains common helper functions, the other methods have been
@@ -312,4 +314,18 @@ public interface ElasticsearchOperations extends DocumentOperations, SearchOpera
 		getIndexOperations().refresh(clazz);
 	}
 	// endregion
+
+	// region helper
+	/**
+	 * gets the String representation for an id.
+	 * 
+	 * @param id
+	 * @return
+	 * @since 4.0
+	 */
+	@Nullable
+	default String stringIdRepresentation(@Nullable Object id) {
+		return Objects.toString(id, null);
+	}
+	// endregion
 }

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

@@ -127,6 +127,7 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 	}
 
 	@Override
+	@Nullable
 	public <T> T get(GetQuery query, Class<T> clazz, IndexCoordinates index) {
 		GetRequest request = requestFactory.getRequest(query, index);
 		try {
@@ -153,12 +154,12 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 	}
 
 	@Override
-	public void bulkIndex(List<IndexQuery> queries, BulkOptions bulkOptions, IndexCoordinates index) {
+	public List<String> bulkIndex(List<IndexQuery> queries, BulkOptions bulkOptions, IndexCoordinates index) {
 
 		Assert.notNull(queries, "List of IndexQuery must not be null");
 		Assert.notNull(bulkOptions, "BulkOptions must not be null");
 
-		doBulkOperation(queries, bulkOptions, index);
+		return doBulkOperation(queries, bulkOptions, index);
 	}
 
 	@Override
@@ -200,10 +201,10 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 		}
 	}
 
-	private void doBulkOperation(List<?> queries, BulkOptions bulkOptions, IndexCoordinates index) {
+	private List<String> doBulkOperation(List<?> queries, BulkOptions bulkOptions, IndexCoordinates index) {
 		BulkRequest bulkRequest = requestFactory.bulkRequest(queries, bulkOptions, index);
 		try {
-			checkForBulkOperationFailure(client.bulk(bulkRequest, RequestOptions.DEFAULT));
+			return checkForBulkOperationFailure(client.bulk(bulkRequest, RequestOptions.DEFAULT));
 		} catch (IOException e) {
 			throw new ElasticsearchException("Error while bulk for request: " + bulkRequest.toString(), e);
 		}
@@ -212,7 +213,7 @@ public class ElasticsearchRestTemplate extends AbstractElasticsearchTemplate {
 
 	// region SearchOperations
 	@Override
-	public long count(Query query,@Nullable Class<?> clazz, IndexCoordinates index) {
+	public long count(Query query, @Nullable Class<?> clazz, IndexCoordinates index) {
 
 		Assert.notNull(query, "query must not be null");
 		Assert.notNull(index, "index must not be null");

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

@@ -127,6 +127,7 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate {
 	}
 
 	@Override
+	@Nullable
 	public <T> T get(GetQuery query, Class<T> clazz, IndexCoordinates index) {
 		GetRequestBuilder getRequestBuilder = requestFactory.getRequestBuilder(client, query, index);
 		GetResponse response = getRequestBuilder.execute().actionGet();
@@ -145,12 +146,12 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate {
 	}
 
 	@Override
-	public void bulkIndex(List<IndexQuery> queries, BulkOptions bulkOptions, IndexCoordinates index) {
+	public List<String> bulkIndex(List<IndexQuery> queries, BulkOptions bulkOptions, IndexCoordinates index) {
 
 		Assert.notNull(queries, "List of IndexQuery must not be null");
 		Assert.notNull(bulkOptions, "BulkOptions must not be null");
 
-		doBulkOperation(queries, bulkOptions, index);
+		return doBulkOperation(queries, bulkOptions, index);
 	}
 
 	@Override
@@ -178,9 +179,9 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate {
 		return updateRequestBuilder.execute().actionGet();
 	}
 
-	private void doBulkOperation(List<?> queries, BulkOptions bulkOptions, IndexCoordinates index) {
+	private List<String> doBulkOperation(List<?> queries, BulkOptions bulkOptions, IndexCoordinates index) {
 		BulkRequestBuilder bulkRequest = requestFactory.bulkRequestBuilder(client, queries, bulkOptions, index);
-		checkForBulkOperationFailure(bulkRequest.execute().actionGet());
+		return checkForBulkOperationFailure(bulkRequest.execute().actionGet());
 	}
 	// endregion
 

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

@@ -176,34 +176,35 @@ public abstract class AbstractElasticsearchRepository<T, ID> implements Elastics
 
 	@Override
 	public <S extends T> S save(S entity) {
+
 		Assert.notNull(entity, "Cannot save 'null' entity.");
-		operations.index(createIndexQuery(entity), getIndexCoordinates());
+
+		operations.save(entity, getIndexCoordinates());
 		indexOperations.refresh(getIndexCoordinates());
 		return entity;
 	}
 
 	public <S extends T> List<S> save(List<S> entities) {
+
 		Assert.notNull(entities, "Cannot insert 'null' as a List.");
+
 		return Streamable.of(saveAll(entities)).stream().collect(Collectors.toList());
 	}
 
 	@Override
 	public <S extends T> S indexWithoutRefresh(S entity) {
 		Assert.notNull(entity, "Cannot save 'null' entity.");
-		operations.index(createIndexQuery(entity), getIndexCoordinates());
+		operations.save(entity);
 		return entity;
 	}
 
 	@Override
 	public <S extends T> Iterable<S> saveAll(Iterable<S> entities) {
-		Assert.notNull(entities, "Cannot insert 'null' as a List.");
-		List<IndexQuery> queries = Streamable.of(entities).stream().map(this::createIndexQuery)
-				.collect(Collectors.toList());
 
-		if (!queries.isEmpty()) {
-			operations.bulkIndex(queries, getIndexCoordinates());
+		Assert.notNull(entities, "Cannot insert 'null' as a List.");
+
+		operations.save(entities, getIndexCoordinates());
 		indexOperations.refresh(getIndexCoordinates());
-		}
 
 		return entities;
 	}
@@ -329,14 +330,6 @@ public abstract class AbstractElasticsearchRepository<T, ID> implements Elastics
 		indexOperations.refresh(getEntityClass());
 	}
 
-	private IndexQuery createIndexQuery(T entity) {
-		IndexQuery query = new IndexQuery();
-		query.setObject(entity);
-		query.setId(stringIdRepresentation(extractIdFromBean(entity)));
-		query.setVersion(extractVersionFromBean(entity));
-		query.setParentId(extractParentIdFromBean(entity));
-		return query;
-	}
 
 	@SuppressWarnings("unchecked")
 	private Class<T> resolveReturnedClassFromGenericType() {
@@ -396,13 +389,6 @@ public abstract class AbstractElasticsearchRepository<T, ID> implements Elastics
 
 	protected abstract @Nullable String stringIdRepresentation(@Nullable ID id);
 
-	private Long extractVersionFromBean(T entity) {
-		return entityInformation.getVersion(entity);
-	}
-
-	private String extractParentIdFromBean(T entity) {
-		return entityInformation.getParentId(entity);
-	}
 
 	private IndexCoordinates getIndexCoordinates() {
 		return operations.getIndexCoordinatesFor(getEntityClass());

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

@@ -15,8 +15,6 @@
  */
 package org.springframework.data.elasticsearch.repository.support;
 
-import java.util.Objects;
-
 import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
 import org.springframework.lang.Nullable;
 
@@ -43,6 +41,6 @@ public class SimpleElasticsearchRepository<T, ID> extends AbstractElasticsearchR
 
 	@Override
 	protected @Nullable String stringIdRepresentation(@Nullable ID id) {
-		return Objects.toString(id, null);
+		return operations.stringIdRepresentation(id);
 	}
 }

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

@@ -2907,6 +2907,84 @@ public abstract class ElasticsearchTemplateTests {
 		assertThat(highlightField.get(1)).contains("<em>message</em>");
 	}
 
+	@Test // DATAES-738
+	void shouldSaveEntityWithIndexCoordinates() {
+		String id = "42";
+		SampleEntity entity = new SampleEntity();
+		entity.setId(id);
+		entity.setVersion(42L);
+		entity.setMessage("message");
+
+		operations.save(entity, index);
+		indexOperations.refresh(index);
+
+		SampleEntity result = operations.get(new GetQuery(id), SampleEntity.class, index);
+
+		assertThat(result).isEqualTo(entity);
+	}
+
+	@Test // DATAES-738
+	void shouldSaveEntityWithOutIndexCoordinates() {
+		String id = "42";
+		SampleEntity entity = new SampleEntity();
+		entity.setId(id);
+		entity.setVersion(42L);
+		entity.setMessage("message");
+
+		operations.save(entity);
+		indexOperations.refresh(index);
+
+		SampleEntity result = operations.get(new GetQuery(id), SampleEntity.class, index);
+
+		assertThat(result).isEqualTo(entity);
+	}
+
+	@Test // DATAES-738
+	void shouldSaveEntityIterableWithIndexCoordinates() {
+		String id1 = "42";
+		SampleEntity entity1 = new SampleEntity();
+		entity1.setId(id1);
+		entity1.setVersion(42L);
+		entity1.setMessage("message");
+		String id2 = "43";
+		SampleEntity entity2 = new SampleEntity();
+		entity2.setId(id2);
+		entity2.setVersion(43L);
+		entity2.setMessage("message");
+
+		operations.save(Arrays.asList(entity1, entity2), index);
+		indexOperations.refresh(index);
+
+		SampleEntity result1 = operations.get(new GetQuery(id1), SampleEntity.class, index);
+		SampleEntity result2 = operations.get(new GetQuery(id2), SampleEntity.class, index);
+
+		assertThat(result1).isEqualTo(entity1);
+		assertThat(result2).isEqualTo(entity2);
+	}
+
+	@Test // DATAES-738
+	void shouldSaveEntityIterableWithoutIndexCoordinates() {
+		String id1 = "42";
+		SampleEntity entity1 = new SampleEntity();
+		entity1.setId(id1);
+		entity1.setVersion(42L);
+		entity1.setMessage("message");
+		String id2 = "43";
+		SampleEntity entity2 = new SampleEntity();
+		entity2.setId(id2);
+		entity2.setVersion(43L);
+		entity2.setMessage("message");
+
+		operations.save(Arrays.asList(entity1, entity2));
+		indexOperations.refresh(index);
+
+		SampleEntity result1 = operations.get(new GetQuery(id1), SampleEntity.class, index);
+		SampleEntity result2 = operations.get(new GetQuery(id2), SampleEntity.class, index);
+
+		assertThat(result1).isEqualTo(entity1);
+		assertThat(result2).isEqualTo(entity2);
+	}
+
 	protected RequestFactory getRequestFactory() {
 		return ((AbstractElasticsearchTemplate) operations).getRequestFactory();
 	}
@@ -3039,8 +3117,7 @@ public abstract class ElasticsearchTemplateTests {
 
 		static class NestedEntity {
 
-			@Nullable
-			@Field(type = Text) private String someField;
+			@Nullable @Field(type = Text) private String someField;
 
 			@Nullable
 			public String getSomeField() {