#220 - Introduce R2dbcEntityTemplate

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-r2dbc</artifactId>
-	<version>1.1.0.BUILD-SNAPSHOT</version>
+	<version>1.1.0.gh-220-SNAPSHOT</version>
 
 	<name>Spring Data R2DBC</name>
 	<description>Spring Data module for R2DBC</description>

src/main/java/org/springframework/data/r2dbc/core/DefaultDatabaseClient.java

@@ -204,6 +204,15 @@ protected Mono<Connection> getConnection() {
 		return ConnectionFactoryUtils.getConnection(obtainConnectionFactory());
 	}
 
+	/**
+	 * Obtain the {@link ReactiveDataAccessStrategy}.
+	 *
+	 * @return a the ReactiveDataAccessStrategy.
+	 */
+	protected ReactiveDataAccessStrategy getDataAccessStrategy() {
+		return dataAccessStrategy;
+	}
+
 	/**
 	 * Release the {@link Connection}.
 	 *
@@ -809,8 +818,8 @@ private <R> FetchSpec<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunctio
 
 			StatementMapper mapper = dataAccessStrategy.getStatementMapper();
 
-			StatementMapper.SelectSpec selectSpec = mapper.createSelect(this.table).withProjection(this.projectedFields)
-					.withSort(this.sort).withPage(this.page);
+			StatementMapper.SelectSpec selectSpec = mapper.createSelect(this.table)
+					.withProjection(this.projectedFields.toArray(new SqlIdentifier[0])).withSort(this.sort).withPage(this.page);
 
 			if (this.criteria != null) {
 				selectSpec = selectSpec.withCriteria(this.criteria);
@@ -922,8 +931,8 @@ private <R> FetchSpec<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunctio
 				columns = this.projectedFields;
 			}
 
-			StatementMapper.SelectSpec selectSpec = mapper.createSelect(this.table).withProjection(columns)
-					.withPage(this.page).withSort(this.sort);
+			StatementMapper.SelectSpec selectSpec = mapper.createSelect(this.table)
+					.withProjection(columns.toArray(new SqlIdentifier[0])).withPage(this.page).withSort(this.sort);
 
 			if (this.criteria != null) {
 				selectSpec = selectSpec.withCriteria(this.criteria);
@@ -1029,7 +1038,7 @@ private <R> FetchSpec<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunctio
 			StatementMapper.InsertSpec insert = mapper.createInsert(this.table);
 
 			for (SqlIdentifier column : this.byName.keySet()) {
-				insert = insert.withColumn(dataAccessStrategy.toSql(column), this.byName.get(column));
+				insert = insert.withColumn(column, this.byName.get(column));
 			}
 
 			PreparedOperation<?> operation = mapper.getMappedObject(insert);
@@ -1152,7 +1161,7 @@ private <MR> FetchSpec<MR> exchange(Object toInsert, BiFunction<Row, RowMetadata
 			for (SqlIdentifier column : outboundRow.keySet()) {
 				SettableValue settableValue = outboundRow.get(column);
 				if (settableValue.hasValue()) {
-					insert = insert.withColumn(dataAccessStrategy.toSql(column), settableValue);
+					insert = insert.withColumn(column, settableValue);
 				}
 			}
 

src/main/java/org/springframework/data/r2dbc/core/DefaultStatementMapper.java

@@ -16,11 +16,8 @@
 package org.springframework.data.r2dbc.core;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 
-import org.springframework.data.domain.Pageable;
-import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.BindTarget;
@@ -84,9 +81,9 @@ public PreparedOperation<?> getMappedObject(SelectSpec selectSpec) {
 	private PreparedOperation<Select> getMappedObject(SelectSpec selectSpec,
 			@Nullable RelationalPersistentEntity<?> entity) {
 
-		Table table = Table.create(toSql(selectSpec.getTable()));
-		List<Column> columns = table.columns(toSql(selectSpec.getProjectedFields()));
-		SelectBuilder.SelectFromAndJoin selectBuilder = StatementBuilder.select(columns).from(table);
+		Table table = selectSpec.getTable();
+		SelectBuilder.SelectFromAndJoin selectBuilder = StatementBuilder.select(getSelectList(selectSpec, entity))
+				.from(table);
 
 		BindMarkers bindMarkers = this.dialect.getBindMarkersFactory().create();
 		Bindings bindings = Bindings.empty();
@@ -102,37 +99,36 @@ private PreparedOperation<Select> getMappedObject(SelectSpec selectSpec,
 
 		if (selectSpec.getSort().isSorted()) {
 
-			Sort mappedSort = this.updateMapper.getMappedObject(selectSpec.getSort(), entity);
-			selectBuilder.orderBy(createOrderByFields(table, mappedSort));
+			List<OrderByField> sort = this.updateMapper.getMappedSort(table, selectSpec.getSort(), entity);
+			selectBuilder.orderBy(sort);
 		}
 
-		if (selectSpec.getPage().isPaged()) {
-
-			Pageable page = selectSpec.getPage();
+		if (selectSpec.getLimit() > 0) {
+			selectBuilder.limit(selectSpec.getLimit());
+		}
 
-			selectBuilder.limitOffset(page.getPageSize(), page.getOffset());
+		if (selectSpec.getOffset() > 0) {
+			selectBuilder.offset(selectSpec.getOffset());
 		}
 
 		Select select = selectBuilder.build();
 		return new DefaultPreparedOperation<>(select, this.renderContext, bindings);
 	}
 
-	private Collection<? extends OrderByField> createOrderByFields(Table table, Sort sortToUse) {
-
-		List<OrderByField> fields = new ArrayList<>();
+	protected List<Expression> getSelectList(SelectSpec selectSpec, @Nullable RelationalPersistentEntity<?> entity) {
 
-		for (Sort.Order order : sortToUse) {
+		if (entity == null) {
+			return selectSpec.getSelectList();
+		}
 
-			OrderByField orderByField = OrderByField.from(table.column(order.getProperty()));
+		List<Expression> selectList = selectSpec.getSelectList();
+		List<Expression> mapped = new ArrayList<>(selectList.size());
 
-			if (order.getDirection() != null) {
-				fields.add(order.isAscending() ? orderByField.asc() : orderByField.desc());
-			} else {
-				fields.add(orderByField);
-			}
+		for (Expression expression : selectList) {
+			mapped.add(updateMapper.getMappedObject(expression, entity));
 		}
 
-		return fields;
+		return mapped;
 	}
 
 	/*

src/main/java/org/springframework/data/r2dbc/core/FluentR2dbcOperations.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2020 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.r2dbc.core;
+
+/**
+ * Stripped down interface providing access to a fluent API that specifies a basic set of reactive R2DBC operations.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ * @see R2dbcEntityOperations
+ */
+public interface FluentR2dbcOperations
+		extends ReactiveSelectOperation, ReactiveInsertOperation, ReactiveUpdateOperation, ReactiveDeleteOperation {}

src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityOperations.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2020 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.r2dbc.core;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.TransientDataAccessResourceException;
+import org.springframework.data.r2dbc.query.Query;
+import org.springframework.data.r2dbc.query.Update;
+
+/**
+ * Interface specifying a basic set of reactive R2DBC operations using entities. Implemented by
+ * {@link R2dbcEntityTemplate}. Not often used directly, but a useful option to enhance testability, as it can easily be
+ * mocked or stubbed.
+ *
+ * @author Mark Paluch
+ * @since 1.1
+ * @see DatabaseClient
+ */
+public interface R2dbcEntityOperations extends FluentR2dbcOperations {
+
+	/**
+	 * Expose the underlying {@link DatabaseClient} to allow SQL operations.
+	 *
+	 * @return the underlying {@link DatabaseClient}.
+	 * @see DatabaseClient
+	 */
+	DatabaseClient getDatabaseClient();
+
+	// -------------------------------------------------------------------------
+	// Methods dealing with org.springframework.data.r2dbc.query.Query
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Returns the number of rows for the given entity class applying {@link Query}. This overridden method allows users
+	 * to further refine the selection Query using a {@link Query} predicate to determine how many entities of the given
+	 * {@link Class type} match the Query.
+	 *
+	 * @param query user-defined count {@link Query} to execute; must not be {@literal null}.
+	 * @param entityClass {@link Class type} of the entity; must not be {@literal null}.
+	 * @return the number of existing entities.
+	 * @throws DataAccessException if any problem occurs while executing the query.
+	 */
+	Mono<Long> count(Query query, Class<?> entityClass) throws DataAccessException;
+
+	/**
+	 * Determine whether the result for {@code entityClass} {@link Query} yields at least one row.
+	 *
+	 * @param query user-defined exists {@link Query} to execute; must not be {@literal null}.
+	 * @param entityClass {@link Class type} of the entity; must not be {@literal null}.
+	 * @return {@literal true} if the object exists.
+	 * @throws DataAccessException if any problem occurs while executing the query.
+	 * @since 2.1
+	 */
+	Mono<Boolean> exists(Query query, Class<?> entityClass) throws DataAccessException;
+
+	/**
+	 * Execute a {@code SELECT} query and convert the resulting items to a stream of entities.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param entityClass The entity type must not be {@literal null}.
+	 * @return the result objects returned by the action.
+	 * @throws DataAccessException if there is any problem issuing the execution.
+	 */
+	<T> Flux<T> select(Query query, Class<T> entityClass) throws DataAccessException;
+
+	/**
+	 * Execute a {@code SELECT} query and convert the resulting item to an entity.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param entityClass The entity type must not be {@literal null}.
+	 * @return the result object returned by the action or {@link Mono#empty()}.
+	 * @throws DataAccessException if there is any problem issuing the execution.
+	 */
+	<T> Mono<T> selectOne(Query query, Class<T> entityClass) throws DataAccessException;
+
+	/**
+	 * Update the queried entities and return {@literal true} if the update was applied.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param update must not be {@literal null}.
+	 * @param entityClass The entity type must not be {@literal null}.
+	 * @return the number of affected rows.
+	 * @throws DataAccessException if there is any problem executing the query.
+	 */
+	Mono<Integer> update(Query query, Update update, Class<?> entityClass) throws DataAccessException;
+
+	/**
+	 * Remove entities (rows)/columns from the table by {@link Query}.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param entityClass The entity type must not be {@literal null}.
+	 * @return the number of affected rows.
+	 * @throws DataAccessException if there is any problem issuing the execution.
+	 */
+	Mono<Integer> delete(Query query, Class<?> entityClass) throws DataAccessException;
+
+	// -------------------------------------------------------------------------
+	// Methods dealing with entities
+	// -------------------------------------------------------------------------
+
+	/**
+	 * Insert the given entity and emit the entity if the insert was applied.
+	 *
+	 * @param entity The entity to insert, must not be {@literal null}.
+	 * @return the inserted entity.
+	 * @throws DataAccessException if there is any problem issuing the execution.
+	 */
+	<T> Mono<T> insert(T entity) throws DataAccessException;
+
+	/**
+	 * Update the given entity and emit the entity if the update was applied.
+	 *
+	 * @param entity The entity to update, must not be {@literal null}.
+	 * @return the updated entity.
+	 * @throws DataAccessException if there is any problem issuing the execution.
+	 * @throws TransientDataAccessResourceException if the update did not affect any rows.
+	 */
+	<T> Mono<T> update(T entity) throws DataAccessException;
+
+	/**
+	 * Delete the given entity and emit the entity if the delete was applied.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @return the deleted entity.
+	 * @throws DataAccessException if there is any problem issuing the execution.
+	 */
+	<T> Mono<T> delete(T entity) throws DataAccessException;
+}