#59 - Consider custom conversion in MappingR2dbcConverter.

MappingR2dbcConverter now considers custom conversions for inbound and outbound conversion of top-level types (Row to Entity, Entity to OutboundRow) and on property level (e.g. convert an object to String and vice versa).

Original pull request: #70.

src/main/asciidoc/reference/mapping.adoc

@@ -28,6 +28,47 @@ Public `JavaBean` properties are not used.
 Otherwise, the zero-argument constructor is used.
 If there is more than one non-zero-argument constructor, an exception will be thrown.
 
+[[mapping-configuration]]
+== Mapping Configuration
+
+Unless explicitly configured, an instance of `MappingR2dbcConverter` is created by default when you create a `DatabaseClient`.
+You can create your own instance of the `MappingR2dbcConverter`.
+By creating your own instance, you can register Spring converters to map specific classes to and from the database.
+
+You can configure the `MappingR2dbcConverter` as well as `DatabaseClient` and `ConnectionFactory` by using  Java-based metadata. The following example uses Spring's Java-based configuration:
+
+.@Configuration class to configure R2DBC mapping support
+====
+[source,java]
+----
+@Configuration
+public class MyAppConfig extends AbstractR2dbcConfiguration {
+
+  public ConnectionFactory connectionFactory() {
+    return ConnectionFactories.get("r2dbc:…");
+  }
+
+  // the following are optional
+
+  @Bean
+  @Override
+  public R2dbcCustomConversions r2dbcCustomConversions() {
+
+    List<Converter<?, ?>> converterList = new ArrayList<Converter<?, ?>>();
+    converterList.add(new org.springframework.data.r2dbc.test.PersonReadConverter());
+    converterList.add(new org.springframework.data.r2dbc.test.PersonWriteConverter());
+    return new R2dbcCustomConversions(getStoreConversions(), converterList);
+  }
+}
+----
+====
+
+`AbstractR2dbcConfiguration` requires you to implement a method that defines a `ConnectionFactory`.
+
+You can add additional converters to the converter by overriding the `r2dbcCustomConversions` method.
+
+NOTE: `AbstractR2dbcConfiguration` creates a `DatabaseClient` instance and registers it with the container under the name `databaseClient`.
+
 [[mapping-usage]]
 == Metadata-based Mapping
 
@@ -52,7 +93,6 @@ public class Person {
 
   private String firstName;
 
-  @Indexed
   private String lastName;
 }
 ----
@@ -103,3 +143,47 @@ class OrderItem {
 
 ----
 
+[[mapping-explicit-converters]]
+=== Overriding Mapping with Explicit Converters
+
+When storing and querying your objects, it is convenient to have a `R2dbcConverter` instance handle the mapping of all Java types to `OutboundRow` instances.
+However, sometimes you may want the `R2dbcConverter` instances do most of the work but let you selectively handle the conversion for a particular type -- perhaps to optimize performance.
+
+To selectively handle the conversion yourself, register one or more one or more `org.springframework.core.convert.converter.Converter` instances with the `R2dbcConverter`.
+
+You can use the `r2dbcCustomConversions` method in `AbstractR2dbcConfiguration` to configure converters. The examples <<mapping-configuration, at the beginning of this chapter>> show how to perform the configuration using Java.
+
+NOTE: Custom top-level entity conversion requires asymmetric types for conversion. Inbound data is extracted from R2DBC's `Row`.
+Outbound data (to be used with `INSERT`/`UPDATE` statements) is represented as `OutboundRow` and later assembled to a statement.
+
+The following example of a Spring Converter implementation converts from a `Row` to a `Person` POJO:
+
+[source,java]
+----
+@ReadingConverter
+ public class PersonReadConverter implements Converter<Row, Person> {
+
+  public Person convert(Row source) {
+    Person p = new Person(source.get("id", String.class),source.get("name", String.class));
+    p.setAge(source.get("age", Integer.class));
+    return p;
+  }
+}
+----
+
+The following example converts from a `Person` to a `OutboundRow`:
+
+[source,java]
+----
+@WritingConverter
+public class PersonWriteConverter implements Converter<Person, OutboundRow> {
+
+  public OutboundRow convert(Person source) {
+    OutboundRow row = new OutboundRow();
+    row.put("_d", source.getId());
+    row.put("name", source.getFirstName());
+    row.put("age", source.getAge());
+    return row;
+  }
+}
+----

src/main/java/org/springframework/data/r2dbc/dialect/R2dbcSimpleTypeHolder.java

@@ -17,10 +17,13 @@ package org.springframework.data.r2dbc.dialect;
 
 import io.r2dbc.spi.Row;
 
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.Set;
 
 import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.data.r2dbc.domain.OutboundRow;
 
 /**
  * Simple constant holder for a {@link SimpleTypeHolder} enriched with R2DBC specific simple types.
@@ -32,7 +35,8 @@ public class R2dbcSimpleTypeHolder extends SimpleTypeHolder {
 	/**
 	 * Set of R2DBC simple types.
 	 */
-	public static final Set<Class<?>> R2DBC_SIMPLE_TYPES = Collections.singleton(Row.class);
+	public static final Set<Class<?>> R2DBC_SIMPLE_TYPES = Collections
+			.unmodifiableSet(new HashSet<>(Arrays.asList(OutboundRow.class, Row.class)));
 
 	public static final SimpleTypeHolder HOLDER = new R2dbcSimpleTypeHolder();
 

src/main/java/org/springframework/data/r2dbc/domain/OutboundRow.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.r2dbc.function.convert;
+package org.springframework.data.r2dbc.domain;
 
 import io.r2dbc.spi.Row;
 

src/main/java/org/springframework/data/r2dbc/domain/SettableValue.java

@@ -13,12 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.r2dbc.function.convert;
+package org.springframework.data.r2dbc.domain;
 
 import java.util.Objects;
 
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 
 /**
  * A database value that can be set in a statement.
@@ -31,13 +32,7 @@ public class SettableValue {
 	private final @Nullable Object value;
 	private final Class<?> type;
 
-	/**
-	 * Create a {@link SettableValue}.
-	 *
-	 * @param value
-	 * @param type
-	 */
-	public SettableValue(@Nullable Object value, Class<?> type) {
+	private SettableValue(@Nullable Object value, Class<?> type) {
 
 		Assert.notNull(type, "Type must not be null");
 
@@ -45,6 +40,42 @@ public class SettableValue {
 		this.type = type;
 	}
 
+	/**
+	 * Creates a new {@link SettableValue} from {@code value}.
+	 *
+	 * @param value must not be {@literal null}.
+	 * @return the {@link SettableValue} value for {@code value}.
+	 */
+	public static SettableValue from(Object value) {
+
+		Assert.notNull(value, "Value must not be null");
+
+		return new SettableValue(value, ClassUtils.getUserClass(value));
+	}
+
+	/**
+	 * Creates a new {@link SettableValue} from {@code value} and {@code type}.
+	 *
+	 * @param value can be {@literal null}.
+	 * @param type must not be {@literal null}.
+	 * @return the {@link SettableValue} value for {@code value}.
+	 */
+	public static SettableValue fromOrEmpty(@Nullable Object value, Class<?> type) {
+		return value == null ? empty(type) : new SettableValue(value, ClassUtils.getUserClass(value));
+	}
+
+	/**
+	 * Creates a new empty {@link SettableValue} for {@code type}.
+	 *
+	 * @return the empty {@link SettableValue} value for {@code type}.
+	 */
+	public static SettableValue empty(Class<?> type) {
+
+		Assert.notNull(type, "Type must not be null");
+
+		return new SettableValue(null, type);
+	}
+
 	/**
 	 * Returns the column value. Can be {@literal null}.
 	 *
@@ -74,6 +105,15 @@ public class SettableValue {
 		return value != null;
 	}
 
+	/**
+	 * Returns whether this {@link SettableValue} has a empty.
+	 *
+	 * @return whether this {@link SettableValue} is empty. {@literal true} if {@link #getValue()} is {@literal null}.
+	 */
+	public boolean isEmpty() {
+		return value == null;
+	}
+
 	@Override
 	public boolean equals(Object o) {
 		if (this == o)
@@ -92,8 +132,8 @@ public class SettableValue {
 	@Override
 	public String toString() {
 		final StringBuffer sb = new StringBuffer();
-		sb.append(getClass().getSimpleName());
-		sb.append(" [value=").append(value);
+		sb.append("SettableValue");
+		sb.append("[value=").append(value);
 		sb.append(", type=").append(type);
 		sb.append(']');
 		return sb.toString();

src/main/java/org/springframework/data/r2dbc/domain/package-info.java

@@ -0,0 +1,7 @@
+/**
+ * Domain objects for R2DBC.
+ */
+@NonNullApi
+package org.springframework.data.r2dbc.domain;
+
+import org.springframework.lang.NonNullApi;

src/main/java/org/springframework/data/r2dbc/function/BindableOperation.java

@@ -2,7 +2,7 @@ package org.springframework.data.r2dbc.function;
 
 import io.r2dbc.spi.Statement;
 
-import org.springframework.data.r2dbc.function.convert.SettableValue;
+import org.springframework.data.r2dbc.domain.SettableValue;
 
 /**
  * Extension to {@link QueryOperation} for operations that allow parameter substitution by binding parameter values.

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

@@ -52,10 +52,10 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.UncategorizedR2dbcException;
+import org.springframework.data.r2dbc.domain.OutboundRow;
+import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.r2dbc.function.connectionfactory.ConnectionProxy;
 import org.springframework.data.r2dbc.function.convert.ColumnMapRowMapper;
-import org.springframework.data.r2dbc.function.convert.OutboundRow;
-import org.springframework.data.r2dbc.function.convert.SettableValue;
 import org.springframework.data.r2dbc.support.R2dbcExceptionTranslator;
 import org.springframework.jdbc.core.SqlProvider;
 import org.springframework.lang.Nullable;
@@ -370,7 +370,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			Assert.notNull(value, () -> String.format("Value at index %d must not be null. Use bindNull(…) instead.", index));
 
 			Map<Integer, SettableValue> byIndex = new LinkedHashMap<>(this.byIndex);
-			byIndex.put(index, new SettableValue(value, value.getClass()));
+			byIndex.put(index, SettableValue.fromOrEmpty(value, value.getClass()));
 
 			return createInstance(byIndex, this.byName, this.sqlSupplier);
 		}
@@ -378,7 +378,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		public ExecuteSpecSupport bindNull(int index, Class<?> type) {
 
 			Map<Integer, SettableValue> byIndex = new LinkedHashMap<>(this.byIndex);
-			byIndex.put(index, new SettableValue(null, type));
+			byIndex.put(index, SettableValue.empty(type));
 
 			return createInstance(byIndex, this.byName, this.sqlSupplier);
 		}
@@ -390,7 +390,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 					() -> String.format("Value for parameter %s must not be null. Use bindNull(…) instead.", name));
 
 			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
-			byName.put(name, new SettableValue(value, value.getClass()));
+			byName.put(name, SettableValue.fromOrEmpty(value, value.getClass()));
 
 			return createInstance(this.byIndex, byName, this.sqlSupplier);
 		}
@@ -400,7 +400,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			Assert.hasText(name, "Parameter name must not be null or empty!");
 
 			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
-			byName.put(name, new SettableValue(null, type));
+			byName.put(name, SettableValue.empty(type));
 
 			return createInstance(this.byIndex, byName, this.sqlSupplier);
 		}
@@ -842,7 +842,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 					() -> String.format("Value for field %s must not be null. Use nullValue(…) instead.", field));
 
 			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
-			byName.put(field, new SettableValue(value, value.getClass()));
+			byName.put(field, SettableValue.fromOrEmpty(value, value.getClass()));
 
 			return new DefaultGenericInsertSpec<>(this.table, byName, this.mappingFunction);
 		}
@@ -853,7 +853,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			Assert.notNull(field, "Field must not be null!");
 
 			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
-			byName.put(field, new SettableValue(null, type));
+			byName.put(field, SettableValue.empty(type));
 
 			return new DefaultGenericInsertSpec<>(this.table, byName, this.mappingFunction);
 		}

src/main/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategy.java

@@ -41,12 +41,12 @@ import org.springframework.data.r2dbc.dialect.BindMarker;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
 import org.springframework.data.r2dbc.dialect.Dialect;
+import org.springframework.data.r2dbc.domain.OutboundRow;
+import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.r2dbc.function.convert.EntityRowMapper;
 import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
-import org.springframework.data.r2dbc.function.convert.OutboundRow;
 import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
 import org.springframework.data.r2dbc.function.convert.R2dbcCustomConversions;
-import org.springframework.data.r2dbc.function.convert.SettableValue;
 import org.springframework.data.r2dbc.support.StatementRenderUtil;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
@@ -182,7 +182,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 					"Dialect " + dialect.getClass().getName() + " does not support array columns");
 		}
 
-		return new SettableValue(converter.getArrayValue(arrayColumns, property, value.getValue()),
+		return SettableValue.fromOrEmpty(converter.getArrayValue(arrayColumns, property, value.getValue()),
 				property.getActualType());
 	}
 

src/main/java/org/springframework/data/r2dbc/function/MapBindParameterSource.java

@@ -18,7 +18,7 @@ package org.springframework.data.r2dbc.function;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
-import org.springframework.data.r2dbc.function.convert.SettableValue;
+import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.util.Assert;
 
 /**
@@ -65,7 +65,7 @@ class MapBindParameterSource implements BindParameterSource {
 		Assert.notNull(paramName, "Parameter name must not be null!");
 		Assert.notNull(value, "Value must not be null!");
 
-		this.values.put(paramName, new SettableValue(value, value.getClass()));
+		this.values.put(paramName, SettableValue.fromOrEmpty(value, value.getClass()));
 		return this;
 	}
 

src/main/java/org/springframework/data/r2dbc/function/ReactiveDataAccessStrategy.java

@@ -26,9 +26,9 @@ import java.util.function.BiFunction;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.r2dbc.dialect.BindMarkersFactory;
-import org.springframework.data.r2dbc.function.convert.OutboundRow;
+import org.springframework.data.r2dbc.domain.OutboundRow;
+import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
-import org.springframework.data.r2dbc.function.convert.SettableValue;
 
 /**
  * Draft of a data access strategy that generalizes convenience operations using mapped entities. Typically used

src/main/java/org/springframework/data/r2dbc/function/convert/MappingR2dbcConverter.java

@@ -25,6 +25,7 @@ import java.lang.reflect.Array;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.BiFunction;
 
 import org.springframework.core.convert.ConversionService;
@@ -38,6 +39,8 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mapping.model.ParameterValueProvider;
 import org.springframework.data.r2dbc.dialect.ArrayColumns;
+import org.springframework.data.r2dbc.domain.OutboundRow;
+import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.relational.core.conversion.BasicRelationalConverter;
 import org.springframework.data.relational.core.conversion.RelationalConverter;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
@@ -135,13 +138,40 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 			}
 
 			Object value = row.get(prefix + property.getColumnName());
-			return readValue(value, property.getTypeInformation());
+			return getPotentiallyConvertedSimpleRead(value, property.getTypeInformation().getType());
 
 		} catch (Exception o_O) {
 			throw new MappingException(String.format("Could not read property %s from result set!", property), o_O);
 		}
 	}
 
+	/**
+	 * Checks whether we have a custom conversion for the given simple object. Converts the given value if so, applies
+	 * {@link Enum} handling or returns the value as is.
+	 *
+	 * @param value
+	 * @param target must not be {@literal null}.
+	 * @return
+	 */
+	@Nullable
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private Object getPotentiallyConvertedSimpleRead(@Nullable Object value, @Nullable Class<?> target) {
+
+		if (value == null || target == null || ClassUtils.isAssignableValue(target, value)) {
+			return value;
+		}
+
+		if (getConversions().hasCustomReadTarget(value.getClass(), target)) {
+			return getConversionService().convert(value, target);
+		}
+
+		if (Enum.class.isAssignableFrom(target)) {
+			return Enum.valueOf((Class<Enum>) target, value.toString());
+		}
+
+		return getConversionService().convert(value, target);
+	}
+
 	private <S> S readEntityFrom(Row row, PersistentProperty<?> property) {
 
 		String prefix = property.getName() + "_";
@@ -182,33 +212,101 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 	public void write(Object source, OutboundRow sink) {
 
 		Class<?> userClass = ClassUtils.getUserClass(source);
-		RelationalPersistentEntity<?> entity = getRequiredPersistentEntity(userClass);
 
-		PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(source);
+		Optional<Class<?>> customTarget = getConversions().getCustomWriteTarget(userClass, OutboundRow.class);
+		if (customTarget.isPresent()) {
+
+			OutboundRow result = getConversionService().convert(source, OutboundRow.class);
+			sink.putAll(result);
+			return;
+		}
+
+		writeInternal(source, sink, userClass);
+	}
+
+	private void writeInternal(Object source, OutboundRow sink, Class<?> userClass) {
+
+		RelationalPersistentEntity<?> entity = getRequiredPersistentEntity(userClass);
+		PersistentPropertyAccessor<?> propertyAccessor = entity.getPropertyAccessor(source);
+
+		writeProperties(sink, entity, propertyAccessor);
+	}
+
+	private void writeProperties(OutboundRow sink, RelationalPersistentEntity<?> entity,
+			PersistentPropertyAccessor<?> accessor) {
 
 		for (RelationalPersistentProperty property : entity) {
 
-			Object writeValue = getWriteValue(propertyAccessor, property);
+			if (!property.isWritable()) {
+				continue;
+			}
 
-			sink.put(property.getColumnName(), new SettableValue(writeValue, property.getType()));
+			Object value = accessor.getProperty(property);
+
+			if (value == null) {
+				writeNullInternal(sink, property);
+				continue;
+			}
+
+			if (!getConversions().isSimpleType(value.getClass())) {
+
+				RelationalPersistentEntity<?> nestedEntity = getMappingContext().getPersistentEntity(property.getActualType());
+				if (nestedEntity != null) {
+					throw new InvalidDataAccessApiUsageException("Nested entities are not supported");
+				}
+			}
+
+			writeSimpleInternal(sink, value, property);
 		}
-
 	}
 
-	@SuppressWarnings("unchecked")
-	private Object getWriteValue(PersistentPropertyAccessor propertyAccessor, RelationalPersistentProperty property) {
+	private void writeSimpleInternal(OutboundRow sink, Object value, RelationalPersistentProperty property) {
+		sink.put(property.getColumnName(), SettableValue.from(getPotentiallyConvertedSimpleWrite(value)));
+	}
 
-		TypeInformation<?> type = property.getTypeInformation();
-		Object value = propertyAccessor.getProperty(property);
+	private void writeNullInternal(OutboundRow sink, RelationalPersistentProperty property) {
 
-		RelationalPersistentEntity<?> nestedEntity = getMappingContext()
-				.getPersistentEntity(type.getRequiredActualType().getType());
+		sink.put(property.getColumnName(),
+				SettableValue.empty(getPotentiallyConvertedSimpleNullType(property.getType())));
+	}
+
+	private Class<?> getPotentiallyConvertedSimpleNullType(Class<?> type) {
+
+		Optional<Class<?>> customTarget = getConversions().getCustomWriteTarget(type);
+
+		if (customTarget.isPresent()) {
+			return customTarget.get();
 
-		if (nestedEntity != null) {
-			throw new InvalidDataAccessApiUsageException("Nested entities are not supported");
 		}
 
-		return value;
+		if (type.isEnum()) {
+			return String.class;
+		}
+
+		return type;
+	}
+
+	/**
+	 * Checks whether we have a custom conversion registered for the given value into an arbitrary simple Mongo type.
+	 * Returns the converted value if so. If not, we perform special enum handling or simply return the value as is.
+	 *
+	 * @param value
+	 * @return
+	 */
+	@Nullable
+	private Object getPotentiallyConvertedSimpleWrite(@Nullable Object value) {
+
+		if (value == null) {
+			return null;
+		}
+
+		Optional<Class<?>> customTarget = getConversions().getCustomWriteTarget(value.getClass());
+
+		if (customTarget.isPresent()) {
+			return getConversionService().convert(value, customTarget.get());
+		}
+
+		return Enum.class.isAssignableFrom(value.getClass()) ? ((Enum<?>) value).name() : value;
 	}
 
 	public Object getArrayValue(ArrayColumns arrayColumns, RelationalPersistentProperty property, Object value) {

src/main/java/org/springframework/data/r2dbc/function/convert/R2dbcConverter.java

@@ -25,6 +25,7 @@ import org.springframework.data.convert.EntityReader;
 import org.springframework.data.convert.EntityWriter;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.r2dbc.dialect.ArrayColumns;
+import org.springframework.data.r2dbc.domain.OutboundRow;
 import org.springframework.data.relational.core.conversion.RelationalConverter;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;

src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java

@@ -30,12 +30,12 @@ import org.reactivestreams.Publisher;
 
 import org.springframework.data.r2dbc.dialect.BindMarker;
 import org.springframework.data.r2dbc.dialect.BindMarkers;
+import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.r2dbc.function.BindIdOperation;
 import org.springframework.data.r2dbc.function.DatabaseClient;
 import org.springframework.data.r2dbc.function.DatabaseClient.GenericExecuteSpec;
 import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
 import org.springframework.data.r2dbc.function.convert.R2dbcConverter;
-import org.springframework.data.r2dbc.function.convert.SettableValue;
 import org.springframework.data.relational.core.sql.Conditions;
 import org.springframework.data.relational.core.sql.Expression;
 import org.springframework.data.relational.core.sql.Functions;

src/test/java/org/springframework/data/r2dbc/domain/SettableValueUnitTests.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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.domain;
+
+import static org.assertj.core.api.Assertions.*;
+
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link SettableValue}.
+ *
+ * @author Mark Paluch
+ */
+public class SettableValueUnitTests {
+
+	@Test // gh-59
+	public void shouldCreateSettableValue() {
+
+		SettableValue value = SettableValue.from("foo");
+
+		assertThat(value.isEmpty()).isFalse();
+		assertThat(value.hasValue()).isTrue();
+		assertThat(value).isEqualTo(SettableValue.from("foo"));
+	}
+
+	@Test // gh-59
+	public void shouldCreateEmpty() {
+
+		SettableValue value = SettableValue.empty(Object.class);
+
+		assertThat(value.isEmpty()).isTrue();
+		assertThat(value.hasValue()).isFalse();
+		assertThat(value).isEqualTo(SettableValue.empty(Object.class));
+		assertThat(value).isNotEqualTo(SettableValue.empty(String.class));
+	}
+
+	@Test // gh-59
+	public void shouldCreatePotentiallyEmpty() {
+
+		assertThat(SettableValue.fromOrEmpty("foo", Object.class).isEmpty()).isFalse();
+		assertThat(SettableValue.fromOrEmpty(null, Object.class).isEmpty()).isTrue();
+	}
+}

src/test/java/org/springframework/data/r2dbc/function/DefaultReactiveDataAccessStrategyUnitTests.java

@@ -14,7 +14,7 @@ import java.util.Map;
 import org.junit.Test;
 
 import org.springframework.data.r2dbc.dialect.PostgresDialect;
-import org.springframework.data.r2dbc.function.convert.SettableValue;
+import org.springframework.data.r2dbc.domain.SettableValue;
 
 /**
  * Unit tests for {@link DefaultReactiveDataAccessStrategy}.

src/test/java/org/springframework/data/r2dbc/function/convert/MappingR2dbcConverterUnitTests.java

@@ -21,9 +21,20 @@ import static org.mockito.Mockito.*;
 import io.r2dbc.spi.Row;
 import lombok.AllArgsConstructor;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.Before;
 import org.junit.Test;
 
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.convert.ReadingConverter;
+import org.springframework.data.convert.WritingConverter;
+import org.springframework.data.r2dbc.domain.OutboundRow;
+import org.springframework.data.r2dbc.domain.SettableValue;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 
 /**
@@ -33,7 +44,21 @@ import org.springframework.data.relational.core.mapping.RelationalMappingContext
  */
 public class MappingR2dbcConverterUnitTests {
 
-	MappingR2dbcConverter converter = new MappingR2dbcConverter(new RelationalMappingContext());
+	RelationalMappingContext mappingContext = new RelationalMappingContext();
+	MappingR2dbcConverter converter = new MappingR2dbcConverter(mappingContext);
+
+	@Before
+	public void before() {
+
+		R2dbcCustomConversions conversions = new R2dbcCustomConversions(
+				Arrays.asList(StringToMapConverter.INSTANCE, MapToStringConverter.INSTANCE,
+						CustomConversionPersonToOutboundRowConverter.INSTANCE, RowToCustomConversionPerson.INSTANCE));
+
+		mappingContext = new RelationalMappingContext();
+		mappingContext.setSimpleTypeHolder(conversions.getSimpleTypeHolder());
+
+		converter = new MappingR2dbcConverter(mappingContext, conversions);
+	}
 
 	@Test // gh-61
 	public void shouldIncludeAllPropertiesInOutboundRow() {
@@ -42,9 +67,9 @@ public class MappingR2dbcConverterUnitTests {
 
 		converter.write(new Person("id", "Walter", "White"), row);
 
-		assertThat(row).containsEntry("id", new SettableValue("id", String.class));
-		assertThat(row).containsEntry("firstname", new SettableValue("Walter", String.class));
-		assertThat(row).containsEntry("lastname", new SettableValue("White", String.class));
+		assertThat(row).containsEntry("id", SettableValue.fromOrEmpty("id", String.class));
+		assertThat(row).containsEntry("firstname", SettableValue.fromOrEmpty("Walter", String.class));
+		assertThat(row).containsEntry("lastname", SettableValue.fromOrEmpty("White", String.class));
 	}
 
 	@Test // gh-41
@@ -68,9 +93,188 @@ public class MappingR2dbcConverterUnitTests {
 		assertThat(result).isEqualTo(42);
 	}
 
+	@Test // gh-59
+	public void shouldFailOnUnsupportedEntity() {
+
+		PersonWithConversions withMap = new PersonWithConversions(null, null, new NonMappableEntity());
+		OutboundRow row = new OutboundRow();
+
+		assertThatThrownBy(() -> converter.write(withMap, row)).isInstanceOf(InvalidDataAccessApiUsageException.class);
+	}
+
+	@Test // gh-59
+	public void shouldConvertMapToString() {
+
+		PersonWithConversions withMap = new PersonWithConversions("foo", Collections.singletonMap("map", "value"), null);
+		OutboundRow row = new OutboundRow();
+		converter.write(withMap, row);
+
+		assertThat(row).containsEntry("nested", SettableValue.from("map"));
+	}
+
+	@Test // gh-59
+	public void shouldReadMapFromString() {
+
+		Row rowMock = mock(Row.class);
+		when(rowMock.get("nested")).thenReturn("map");
+
+		PersonWithConversions result = converter.read(PersonWithConversions.class, rowMock);
+
+		assertThat(result.nested).isEqualTo(Collections.singletonMap("map", "map"));
+	}
+
+	@Test // gh-59
+	public void shouldConvertEnum() {
+
+		WithEnum withMap = new WithEnum("foo", Condition.Mint);
+		OutboundRow row = new OutboundRow();
+		converter.write(withMap, row);
+
+		assertThat(row).containsEntry("condition", SettableValue.from("Mint"));
+	}
+
+	@Test // gh-59
+	public void shouldConvertNullEnum() {
+
+		WithEnum withMap = new WithEnum("foo", null);
+		OutboundRow row = new OutboundRow();
+		converter.write(withMap, row);
+
+		assertThat(row).containsEntry("condition", SettableValue.fromOrEmpty(null, String.class));
+	}
+
+	@Test // gh-59
+	public void shouldReadEnum() {
+
+		Row rowMock = mock(Row.class);
+		when(rowMock.get("condition")).thenReturn("Mint");
+
+		WithEnum result = converter.read(WithEnum.class, rowMock);
+
+		assertThat(result.condition).isEqualTo(Condition.Mint);
+	}
+
+	@Test // gh-59
+	public void shouldWriteTopLevelEntity() {
+
+		CustomConversionPerson person = new CustomConversionPerson();
+		person.entity = new NonMappableEntity();
+		person.foo = "bar";
+
+		OutboundRow row = new OutboundRow();
+		converter.write(person, row);
+
+		assertThat(row).containsEntry("foo_column", SettableValue.from("bar")).containsEntry("entity",
+				SettableValue.from("nested_entity"));
+	}
+
+	@Test // gh-59
+	public void shouldReadTopLevelEntity() {
+
+		Row rowMock = mock(Row.class);
+		when(rowMock.get("foo_column", String.class)).thenReturn("bar");
+		when(rowMock.get("nested_entity")).thenReturn("map");
+
+		CustomConversionPerson result = converter.read(CustomConversionPerson.class, rowMock);
+
+		assertThat(result.foo).isEqualTo("bar");
+		assertThat(result.entity).isNotNull();
+	}
+
 	@AllArgsConstructor
 	static class Person {
 		@Id String id;
 		String firstname, lastname;
 	}
+
+	@AllArgsConstructor
+	static class WithEnum {
+		@Id String id;
+		Condition condition;
+	}
+
+	enum Condition {
+		Mint, Used
+	}
+
+	@AllArgsConstructor
+	static class PersonWithConversions {
+		@Id String id;
+		Map<String, String> nested;
+		NonMappableEntity unsupported;
+	}
+
+	static class CustomConversionPerson {
+
+		String foo;
+		NonMappableEntity entity;
+	}
+
+	static class NonMappableEntity {}
+
+	@ReadingConverter
+	enum StringToMapConverter implements Converter<String, Map<String, String>> {
+
+		INSTANCE;
+
+		@Override
+		public Map<String, String> convert(String source) {
+
+			if (source != null) {
+				return Collections.singletonMap(source, source);
+			}
+
+			return null;
+		}
+	}
+
+	@WritingConverter
+	enum MapToStringConverter implements Converter<Map<String, String>, String> {
+
+		INSTANCE;
+
+		@Override
+		public String convert(Map<String, String> source) {
+
+			if (!source.isEmpty()) {
+				return source.keySet().iterator().next();
+			}
+
+			return null;
+		}
+	}
+
+	@WritingConverter
+	enum CustomConversionPersonToOutboundRowConverter implements Converter<CustomConversionPerson, OutboundRow> {
+
+		INSTANCE;
+
+		@Override
+		public OutboundRow convert(CustomConversionPerson source) {
+
+			OutboundRow row = new OutboundRow();
+			row.put("foo_column", SettableValue.from(source.foo));
+			row.put("entity", SettableValue.from("nested_entity"));
+
+			return row;
+		}
+	}
+
+	@ReadingConverter
+	enum RowToCustomConversionPerson implements Converter<Row, CustomConversionPerson> {
+
+		INSTANCE;
+
+		@Override
+		public CustomConversionPerson convert(Row source) {
+
+			CustomConversionPerson person = new CustomConversionPerson();
+			person.foo = source.get("foo_column", String.class);
+
+			Object nested_entity = source.get("nested_entity");
+			person.entity = nested_entity != null ? new NonMappableEntity() : null;
+
+			return person;
+		}
+	}
 }