Fix test

experiment with resolving reactive types in document

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>4.1.0-SNAPSHOT</version>
+	<version>4.1.x-MANUAL-ENCRYPTION-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.1.0-SNAPSHOT</version>
+		<version>4.1.x-MANUAL-ENCRYPTION-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-distribution/pom.xml

@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.1.0-SNAPSHOT</version>
+		<version>4.1.x-MANUAL-ENCRYPTION-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -13,7 +13,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.1.0-SNAPSHOT</version>
+		<version>4.1.x-MANUAL-ENCRYPTION-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 
@@ -112,6 +112,13 @@
 			<optional>true</optional>
 		</dependency>
 
+		<dependency>
+			<groupId>org.mongodb</groupId>
+			<artifactId>mongodb-crypt</artifactId>
+			<version>1.6.1</version>
+			<optional>true</optional>
+		</dependency>
+
 		<dependency>
 			<groupId>io.projectreactor</groupId>
 			<artifactId>reactor-core</artifactId>

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoExceptionTranslator.java

@@ -68,6 +68,8 @@ public class MongoExceptionTranslator implements PersistenceExceptionTranslator
 	private static final Set<String> DATA_INTEGRITY_EXCEPTIONS = new HashSet<>(
 			Arrays.asList("WriteConcernException", "MongoWriteException", "MongoBulkWriteException"));
 
+	private static final Set<String> SECURITY_EXCEPTIONS = Set.of("MongoCryptException");
+
 	@Nullable
 	public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
 
@@ -131,6 +133,8 @@ public class MongoExceptionTranslator implements PersistenceExceptionTranslator
 				return new ClientSessionException(ex.getMessage(), ex);
 			} else if (MongoDbErrorCodes.isTransactionFailureCode(code)) {
 				return new MongoTransactionException(ex.getMessage(), ex);
+			} else if(ex.getCause() != null && SECURITY_EXCEPTIONS.contains(ClassUtils.getShortName(ex.getCause().getClass()))) {
+				return new PermissionDeniedDataAccessException(ex.getMessage(), ex);
 			}
 
 			return new UncategorizedMongoDbException(ex.getMessage(), ex);

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -30,6 +30,7 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
@@ -1431,7 +1432,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 				Document dbDoc = entity.toMappedDocument(writer).getDocument();
 				maybeEmitEvent(new BeforeSaveEvent<T>(toConvert, dbDoc, collectionName));
 
-				return maybeCallBeforeSave(toConvert, dbDoc, collectionName).flatMap(it -> {
+				return maybeCallBeforeSave(toConvert, dbDoc, collectionName)
+						.flatMap(it -> {
 
 					return saveDocument(collectionName, dbDoc, it.getClass()).flatMap(id -> {
 
@@ -1444,6 +1446,26 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		});
 	}
 
+	private Mono<Document> resolveValues(Mono<Document> document) {
+		return document.flatMap(source -> {
+			for (Entry<String, Object> entry : source.entrySet()) {
+				if (entry.getValue()instanceof Mono<?> valueMono) {
+					return valueMono.flatMap(value -> {
+						source.put(entry.getKey(), value);
+						return resolveValues(Mono.just(source));
+					});
+				}
+				if (entry.getValue()instanceof Document nested) {
+					return resolveValues(Mono.just(nested)).map(it -> {
+						source.put(entry.getKey(), it);
+						return source;
+					});
+				}
+			}
+			return Mono.just(source);
+		});
+	}
+
 	protected Mono<Object> insertDocument(String collectionName, Document dbDoc, Class<?> entityClass) {
 
 		if (LOGGER.isDebugEnabled()) {
@@ -1527,16 +1549,16 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 					? collection //
 					: collection.withWriteConcern(writeConcernToUse);
 
-			Publisher<?> publisher;
+			Publisher<?> publisher = null;
+			Mono<Document> resolved = resolveValues(Mono.just(queryOperations.createInsertContext(mapped).prepareId(entityClass).getDocument()));
 			if (!mapped.hasId()) {
-				publisher = collectionToUse
-						.insertOne(queryOperations.createInsertContext(mapped).prepareId(entityClass).getDocument());
+				publisher = resolved.flatMap(it -> Mono.from(collectionToUse.insertOne(it)));
 			} else {
 
 				MongoPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityClass);
 				UpdateContext updateContext = queryOperations.replaceSingleContext(mapped, true);
 				Document filter = updateContext.getMappedQuery(entity);
-				Document replacement = updateContext.getMappedUpdate(entity);
+				Mono<Document> replacement = resolveValues(Mono.just(updateContext.getMappedUpdate(entity)));
 
 				Mono<Document> deferredFilter;
 
@@ -1547,14 +1569,17 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 						deferredFilter = Mono
 								.from(
 										collection.find(filter, Document.class).projection(updateContext.getMappedShardKey(entity)).first())
-								.defaultIfEmpty(replacement).map(it -> updateContext.applyShardKey(entity, filter, it));
+								.switchIfEmpty(replacement)
+								.map(it -> {
+									return updateContext.applyShardKey(entity, filter, it);
+								});
 					}
 				} else {
 					deferredFilter = Mono.just(filter);
 				}
 
-				publisher = deferredFilter.flatMapMany(
-						it -> collectionToUse.replaceOne(it, replacement, updateContext.getReplaceOptions(entityClass)));
+				publisher = deferredFilter.zipWith(replacement).flatMapMany(
+						it -> collectionToUse.replaceOne(it.getT1(), it.getT2(), updateContext.getReplaceOptions(entityClass)));
 			}
 
 			return Mono.from(publisher).map(o -> mapped.getId());

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MappingMongoConverter.java

@@ -868,9 +868,9 @@ public class MappingMongoConverter extends AbstractMongoConverter implements App
 					dbObjectAccessor.put(prop, null);
 				}
 			} else if (!conversions.isSimpleType(value.getClass())) {
-				writePropertyInternal(value, dbObjectAccessor, prop);
+				writePropertyInternal(value, dbObjectAccessor, prop, accessor);
 			} else {
-				writeSimpleInternal(value, bson, prop);
+				writeSimpleInternal(value, bson, prop, accessor);
 			}
 		}
 	}
@@ -887,11 +887,11 @@ public class MappingMongoConverter extends AbstractMongoConverter implements App
 			return;
 		}
 
-		writePropertyInternal(value, dbObjectAccessor, inverseProp);
+		writePropertyInternal(value, dbObjectAccessor, inverseProp, accessor);
 	}
 
 	@SuppressWarnings({ "unchecked" })
-	protected void writePropertyInternal(@Nullable Object obj, DocumentAccessor accessor, MongoPersistentProperty prop) {
+	protected void writePropertyInternal(@Nullable Object obj, DocumentAccessor accessor, MongoPersistentProperty prop, PersistentPropertyAccessor<?> persistentPropertyAccessor) {
 
 		if (obj == null) {
 			return;
@@ -902,7 +902,13 @@ public class MappingMongoConverter extends AbstractMongoConverter implements App
 
 		if (conversions.hasValueConverter(prop)) {
 			accessor.put(prop, conversions.getPropertyValueConversions().getValueConverter(prop).write(obj,
-					new MongoConversionContext(prop, this)));
+					new MongoConversionContext(new PropertyValueProvider<MongoPersistentProperty>() {
+						@Nullable
+						@Override
+						public <T> T getPropertyValue(MongoPersistentProperty property) {
+							return (T) persistentPropertyAccessor.getProperty(property);
+						}
+					}, prop, this)));
 			return;
 		}
 
@@ -1234,12 +1240,18 @@ public class MappingMongoConverter extends AbstractMongoConverter implements App
 		BsonUtils.addToMap(bson, key, getPotentiallyConvertedSimpleWrite(value, Object.class));
 	}
 
-	private void writeSimpleInternal(@Nullable Object value, Bson bson, MongoPersistentProperty property) {
+	private void writeSimpleInternal(@Nullable Object value, Bson bson, MongoPersistentProperty property, PersistentPropertyAccessor<?> persistentPropertyAccessor) {
 		DocumentAccessor accessor = new DocumentAccessor(bson);
 
 		if (conversions.hasValueConverter(property)) {
 			accessor.put(property, conversions.getPropertyValueConversions().getValueConverter(property).write(value,
-					new MongoConversionContext(property, this)));
+					new MongoConversionContext(new PropertyValueProvider<MongoPersistentProperty>() {
+						@Nullable
+						@Override
+						public <T> T getPropertyValue(MongoPersistentProperty property) {
+							return (T) persistentPropertyAccessor.getProperty(property);
+						}
+					}, property, this)));
 			return;
 		}
 
@@ -1892,7 +1904,7 @@ public class MappingMongoConverter extends AbstractMongoConverter implements App
 			CustomConversions conversions = context.getCustomConversions();
 			if (conversions.hasValueConverter(property)) {
 				return (T) conversions.getPropertyValueConversions().getValueConverter(property).read(value,
-						new MongoConversionContext(property, context.getSourceConverter()));
+						new MongoConversionContext(this, property, context.getSourceConverter()));
 			}
 
 			ConversionContext contextToUse = context.forProperty(property);

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConversionContext.java

@@ -17,6 +17,8 @@ package org.springframework.data.mongodb.core.convert;
 
 import org.bson.conversions.Bson;
 import org.springframework.data.convert.ValueConversionContext;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.model.PropertyValueProvider;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
@@ -29,11 +31,13 @@ import org.springframework.lang.Nullable;
  */
 public class MongoConversionContext implements ValueConversionContext<MongoPersistentProperty> {
 
+	private final PropertyValueProvider accessor; // TODO: generics
 	private final MongoPersistentProperty persistentProperty;
 	private final MongoConverter mongoConverter;
 
-	public MongoConversionContext(MongoPersistentProperty persistentProperty, MongoConverter mongoConverter) {
+	public MongoConversionContext(PropertyValueProvider<?> accessor, MongoPersistentProperty persistentProperty, MongoConverter mongoConverter) {
 
+		this.accessor = accessor;
 		this.persistentProperty = persistentProperty;
 		this.mongoConverter = mongoConverter;
 	}
@@ -43,6 +47,10 @@ public class MongoConversionContext implements ValueConversionContext<MongoPersi
 		return persistentProperty;
 	}
 
+	public Object getValue(String propertyPath) {
+		return accessor.getPropertyValue(persistentProperty.getOwner().getRequiredPersistentProperty(propertyPath));
+	}
+
 	@Override
 	public <T> T write(@Nullable Object value, TypeInformation<T> target) {
 		return (T) mongoConverter.convertToMongoType(value, target);

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/QueryMapper.java

@@ -437,7 +437,7 @@ public class QueryMapper {
 				&& converter.getCustomConversions().hasValueConverter(documentField.getProperty())) {
 			return converter.getCustomConversions().getPropertyValueConversions()
 					.getValueConverter(documentField.getProperty())
-					.write(value, new MongoConversionContext(documentField.getProperty(), converter));
+					.write(value, new MongoConversionContext(null, documentField.getProperty(), converter));
 		}
 
 		if (documentField.isIdField() && !documentField.isAssociation()) {

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/convert/MappingMongoConverterUnitTests.java

@@ -2613,7 +2613,7 @@ class MappingMongoConverterUnitTests {
 		doReturn(Person.class).when(persistentProperty).getType();
 		doReturn(Person.class).when(persistentProperty).getRawType();
 
-		converter.writePropertyInternal(sourceValue, accessor, persistentProperty);
+		converter.writePropertyInternal(sourceValue, accessor, persistentProperty, null);
 
 		assertThat(accessor.getDocument())
 				.isEqualTo(new org.bson.Document("pName", new org.bson.Document("_id", id.toString())));

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/fle/FLETests.java

@@ -0,0 +1,528 @@
+/*
+ * Copyright 2022 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.mongodb.fle;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.mongodb.core.EncryptionAlgorithms.*;
+import static org.springframework.data.mongodb.core.query.Criteria.*;
+
+import lombok.Data;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.security.SecureRandom;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.bson.BsonArray;
+import org.bson.BsonBinary;
+import org.bson.BsonDocument;
+import org.bson.BsonValue;
+import org.bson.Document;
+import org.bson.types.Binary;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.CollectionFactory;
+import org.springframework.core.annotation.AliasFor;
+import org.springframework.dao.PermissionDeniedDataAccessException;
+import org.springframework.data.convert.PropertyValueConverterFactory;
+import org.springframework.data.convert.ValueConverter;
+import org.springframework.data.mongodb.config.AbstractMongoClientConfiguration;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.convert.MongoConversionContext;
+import org.springframework.data.mongodb.core.convert.MongoCustomConversions.MongoConverterConfigurationAdapter;
+import org.springframework.data.mongodb.core.convert.MongoValueConverter;
+import org.springframework.data.mongodb.core.mapping.Encrypted;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
+import org.springframework.data.mongodb.core.query.Update;
+import org.springframework.data.mongodb.fle.FLETests.Config;
+import org.springframework.data.mongodb.util.BsonUtils;
+import org.springframework.data.util.Lazy;
+import org.springframework.lang.Nullable;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
+
+import com.mongodb.ClientEncryptionSettings;
+import com.mongodb.ConnectionString;
+import com.mongodb.MongoClientSettings;
+import com.mongodb.MongoNamespace;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.model.Filters;
+import com.mongodb.client.model.IndexOptions;
+import com.mongodb.client.model.Indexes;
+import com.mongodb.client.model.vault.DataKeyOptions;
+import com.mongodb.client.model.vault.EncryptOptions;
+import com.mongodb.client.result.DeleteResult;
+import com.mongodb.client.vault.ClientEncryption;
+import com.mongodb.client.vault.ClientEncryptions;
+
+/**
+ * @author Christoph Strobl
+ */
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration(classes = Config.class)
+public class FLETests {
+
+	@Autowired MongoTemplate template;
+
+	@Test
+	void manualEnAndDecryption() {
+
+		Person person = new Person();
+		person.id = "id-1";
+		person.name = "p1-name";
+		person.ssn = "mySecretSSN"; // determinisitc encryption (queryable)
+		person.wallet = "myEvenMoreSecretStuff"; // random encryption (non queryable)
+
+		// nested full document encryption
+		person.address = new Address();
+		person.address.city = "NYC";
+		person.address.street = "4th Ave.";
+
+		person.encryptedZip = new AddressWithEncryptedZip();
+		person.encryptedZip.city = "Boston";
+		person.encryptedZip.street = "central square";
+		person.encryptedZip.zip = "1234567890";
+
+		person.listOfString = Arrays.asList("spring", "data", "mongodb");
+
+		Address partOfList = new Address();
+		partOfList.city = "SFO";
+		partOfList.street = "---";
+		person.listOfComplex = Collections.singletonList(partOfList);
+
+		template.save(person);
+
+		System.out.println("source: " + person);
+
+		Document savedDocument = template.execute(Person.class, collection -> {
+			return collection.find(new Document()).first();
+		});
+
+		// ssn should look like "ssn": {"$binary": {"base64": "...
+		System.out.println("saved: " + savedDocument.toJson());
+		assertThat(savedDocument.get("ssn")).isInstanceOf(Binary.class);
+		assertThat(savedDocument.get("wallet")).isInstanceOf(Binary.class);
+		assertThat(savedDocument.get("encryptedZip")).isInstanceOf(Document.class);
+		assertThat(savedDocument.get("encryptedZip", Document.class).get("zip")).isInstanceOf(Binary.class);
+		assertThat(savedDocument.get("address")).isInstanceOf(Binary.class);
+		assertThat(savedDocument.get("listOfString")).isInstanceOf(Binary.class);
+		assertThat(savedDocument.get("listOfComplex")).isInstanceOf(Binary.class);
+
+		// count should be 1 using a deterministic algorithm
+		long queryCount = template.query(Person.class).matching(where("ssn").is(person.ssn)).count();
+		System.out.println("query(count): " + queryCount);
+		assertThat(queryCount).isOne();
+
+		Person bySsn = template.query(Person.class).matching(where("ssn").is(person.ssn)).firstValue();
+		System.out.println("queryable: " + bySsn);
+		assertThat(bySsn).isEqualTo(person);
+
+		Person byWallet = template.query(Person.class).matching(where("wallet").is(person.wallet)).firstValue();
+		System.out.println("not-queryable: " + byWallet);
+		assertThat(byWallet).isNull();
+	}
+
+	@Test
+	void theUpdateStuff() {
+
+		Person person = new Person();
+		person.id = "id-1";
+		person.name = "p1-name";
+
+		template.save(person);
+
+		Document savedDocument = template.execute(Person.class, collection -> {
+			return collection.find(new Document()).first();
+		});
+		System.out.println("saved: " + savedDocument.toJson());
+
+		template.update(Person.class).matching(where("id").is(person.id)).apply(Update.update("ssn", "secret-value")).first();
+
+		savedDocument = template.execute(Person.class, collection -> {
+			return collection.find(new Document()).first();
+		});
+		System.out.println("updated: " + savedDocument.toJson());
+		assertThat(savedDocument.get("ssn")).isInstanceOf(Binary.class);
+
+	}
+
+	@Test
+	void altKeyDetection(@Autowired ClientEncryption clientEncryption) throws InterruptedException {
+
+		BsonBinary user1key = clientEncryption.createDataKey("local",
+				new DataKeyOptions().keyAltNames(Collections.singletonList("user-1")));
+
+		BsonBinary user2key = clientEncryption.createDataKey("local",
+				new DataKeyOptions().keyAltNames(Collections.singletonList("user-2")));
+
+		Person p1 = new Person();
+		p1.id = "id-1";
+		p1.name = "user-1";
+		p1.ssn = "ssn";
+		p1.viaAltKeyNameField = "value-1";
+
+		Person p2 = new Person();
+		p2.id = "id-2";
+		p2.name = "user-2";
+		p2.viaAltKeyNameField = "value-1";
+
+		Person p3 = new Person();
+		p3.id = "id-3";
+		p3.name = "user-1";
+		p3.viaAltKeyNameField = "value-1";
+
+		template.save(p1);
+		template.save(p2);
+		template.save(p3);
+
+		template.execute(Person.class, collection -> {
+			collection.find(new Document()).forEach(it -> System.out.println(it.toJson()));
+			return null;
+		});
+
+		// System.out.println(template.query(Person.class).matching(where("id").is(p1.id)).firstValue());
+		// System.out.println(template.query(Person.class).matching(where("id").is(p2.id)).firstValue());
+
+		DeleteResult deleteResult = clientEncryption.deleteKey(user2key);
+		clientEncryption.getKeys().forEach(System.out::println);
+		System.out.println("deleteResult: " + deleteResult);
+
+		System.out.println("---- waiting for cache timeout ----");
+		TimeUnit.SECONDS.sleep(90);
+
+		assertThat(template.query(Person.class).matching(where("id").is(p1.id)).firstValue()).isEqualTo(p1);
+
+		assertThatExceptionOfType(PermissionDeniedDataAccessException.class)
+				.isThrownBy(() -> template.query(Person.class).matching(where("id").is(p2.id)).firstValue());
+	}
+
+	@Configuration
+	static class Config extends AbstractMongoClientConfiguration {
+
+		@Autowired ApplicationContext applicationContext;
+
+		@Override
+		protected String getDatabaseName() {
+			return "fle-test";
+		}
+
+		@Bean
+		public MongoClient mongoClient() {
+			return super.mongoClient();
+		}
+
+		@Override
+		protected void configureConverters(MongoConverterConfigurationAdapter converterConfigurationAdapter) {
+
+			converterConfigurationAdapter
+					.registerPropertyValueConverterFactory(PropertyValueConverterFactory.beanFactoryAware(applicationContext));
+		}
+
+		@Bean
+		EncryptingConverter encryptingConverter(ClientEncryption clientEncryption) {
+			return new EncryptingConverter(clientEncryption);
+		}
+
+		@Bean
+		ClientEncryption clientEncryption(MongoClient mongoClient) {
+
+			final byte[] localMasterKey = new byte[96];
+			new SecureRandom().nextBytes(localMasterKey);
+			Map<String, Map<String, Object>> kmsProviders = new HashMap<String, Map<String, Object>>() {
+				{
+					put("local", new HashMap<String, Object>() {
+						{
+							put("key", localMasterKey);
+						}
+					});
+				}
+			};
+
+			MongoNamespace keyVaultNamespace = new MongoNamespace("encryption.testKeyVault");
+			MongoCollection<Document> keyVaultCollection = mongoClient.getDatabase(keyVaultNamespace.getDatabaseName())
+					.getCollection(keyVaultNamespace.getCollectionName());
+			keyVaultCollection.drop();
+			// Ensure that two data keys cannot share the same keyAltName.
+			keyVaultCollection.createIndex(Indexes.ascending("keyAltNames"),
+					new IndexOptions().unique(true).partialFilterExpression(Filters.exists("keyAltNames")));
+
+			MongoCollection<Document> collection = mongoClient.getDatabase(getDatabaseName()).getCollection("test");
+			collection.drop(); // Clear old data
+
+			// Create the ClientEncryption instance
+			ClientEncryptionSettings clientEncryptionSettings = ClientEncryptionSettings.builder()
+					.keyVaultMongoClientSettings(
+							MongoClientSettings.builder().applyConnectionString(new ConnectionString("mongodb://localhost")).build())
+					.keyVaultNamespace(keyVaultNamespace.getFullName()).kmsProviders(kmsProviders).build();
+			ClientEncryption clientEncryption = ClientEncryptions.create(clientEncryptionSettings);
+			return clientEncryption;
+		}
+	}
+
+	@Data
+	@org.springframework.data.mongodb.core.mapping.Document("test")
+	static class Person {
+
+		String id;
+		String name;
+
+		@EncryptedField(algorithm = AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic) //
+		String ssn;
+
+		@EncryptedField(algorithm = AEAD_AES_256_CBC_HMAC_SHA_512_Random, altKeyName = "mySuperSecretKey") //
+		String wallet;
+
+		@EncryptedField(algorithm = AEAD_AES_256_CBC_HMAC_SHA_512_Random) // full document must be random
+		Address address;
+
+		AddressWithEncryptedZip encryptedZip;
+
+		@EncryptedField(algorithm = AEAD_AES_256_CBC_HMAC_SHA_512_Random) // lists must be random
+		List<String> listOfString;
+
+		@EncryptedField(algorithm = AEAD_AES_256_CBC_HMAC_SHA_512_Random) // lists must be random
+		List<Address> listOfComplex;
+
+		@EncryptedField(algorithm = AEAD_AES_256_CBC_HMAC_SHA_512_Random, altKeyName = "/name") //
+		String viaAltKeyNameField;
+	}
+
+	@Data
+	static class Address {
+		String city;
+		String street;
+	}
+
+	@Getter
+	@Setter
+	static class AddressWithEncryptedZip extends Address {
+
+		@EncryptedField(algorithm = AEAD_AES_256_CBC_HMAC_SHA_512_Random) String zip;
+
+		@Override
+		public String toString() {
+			return "AddressWithEncryptedZip{" + "zip='" + zip + '\'' + ", city='" + getCity() + '\'' + ", street='"
+					+ getStreet() + '\'' + '}';
+		}
+	}
+
+	@Retention(RetentionPolicy.RUNTIME)
+	@Target(ElementType.FIELD)
+	@Encrypted
+	@ValueConverter(EncryptingConverter.class)
+	@interface EncryptedField {
+
+		@AliasFor(annotation = Encrypted.class, value = "algorithm")
+		String algorithm() default "";
+
+		String altKeyName() default "";
+	}
+
+	static class EncryptingConverter implements MongoValueConverter<Object, Object> {
+
+		private ClientEncryption clientEncryption;
+		private BsonBinary dataKeyId; // should be provided from outside.
+
+		public EncryptingConverter(ClientEncryption clientEncryption) {
+
+			this.clientEncryption = clientEncryption;
+			this.dataKeyId = clientEncryption.createDataKey("local",
+					new DataKeyOptions().keyAltNames(Collections.singletonList("mySuperSecretKey")));
+		}
+
+		@Nullable
+		@Override
+		public Object read(Object value, MongoConversionContext context) {
+
+			ManualEncryptionContext encryptionContext = buildEncryptionContext(context);
+			Object decrypted = encryptionContext.decrypt(value, clientEncryption);
+			return decrypted instanceof BsonValue ? BsonUtils.toJavaType((BsonValue) decrypted) : decrypted;
+		}
+
+		@Nullable
+		@Override
+		public BsonBinary write(Object value, MongoConversionContext context) {
+
+			ManualEncryptionContext encryptionContext = buildEncryptionContext(context);
+			return encryptionContext.encrypt(value, clientEncryption);
+		}
+
+		ManualEncryptionContext buildEncryptionContext(MongoConversionContext context) {
+			return new ManualEncryptionContext(context, this.dataKeyId);
+		}
+	}
+
+	static class ManualEncryptionContext {
+
+		MongoConversionContext context;
+		MongoPersistentProperty persistentProperty;
+		BsonBinary dataKeyId;
+		Lazy<Encrypted> encryption;
+
+		public ManualEncryptionContext(MongoConversionContext context, BsonBinary dataKeyId) {
+			this.context = context;
+			this.persistentProperty = context.getProperty();
+			this.dataKeyId = dataKeyId;
+			this.encryption = Lazy.of(() -> persistentProperty.findAnnotation(Encrypted.class));
+		}
+
+		BsonBinary encrypt(Object value, ClientEncryption clientEncryption) {
+
+			// TODO: check - encryption.get().keyId()
+
+			EncryptOptions encryptOptions = new EncryptOptions(encryption.get().algorithm());
+
+			EncryptedField annotation = persistentProperty.findAnnotation(EncryptedField.class);
+			if (annotation != null && !annotation.altKeyName().isBlank()) {
+				if (annotation.altKeyName().startsWith("/")) {
+					String fieldName = annotation.altKeyName().replace("/", "");
+					Object altKeyNameValue = context.getValue(fieldName);
+					encryptOptions = encryptOptions.keyAltName(altKeyNameValue.toString());
+				} else {
+					encryptOptions = encryptOptions.keyAltName(annotation.altKeyName());
+				}
+			} else {
+				encryptOptions = encryptOptions.keyId(this.dataKeyId);
+			}
+
+			System.out.println(
+					"encrypting with: " + (StringUtils.hasText(encryptOptions.getKeyAltName()) ? encryptOptions.getKeyAltName()
+							: encryptOptions.getKeyId()));
+
+			if (!persistentProperty.isEntity()) {
+
+				if (persistentProperty.isCollectionLike()) {
+					return clientEncryption.encrypt(collectionLikeToBsonValue(value), encryptOptions);
+				}
+				return clientEncryption.encrypt(BsonUtils.simpleToBsonValue(value), encryptOptions);
+			}
+			if (persistentProperty.isCollectionLike()) {
+				return clientEncryption.encrypt(collectionLikeToBsonValue(value), encryptOptions);
+			}
+
+			Object write = context.write(value);
+			if (write instanceof Document doc) {
+				return clientEncryption.encrypt(doc.toBsonDocument(), encryptOptions);
+			}
+			return clientEncryption.encrypt(BsonUtils.simpleToBsonValue(write), encryptOptions);
+		}
+
+		public BsonValue collectionLikeToBsonValue(Object value) {
+
+			if (persistentProperty.isCollectionLike()) {
+
+				BsonArray bsonArray = new BsonArray();
+				if (!persistentProperty.isEntity()) {
+					if (value instanceof Collection values) {
+						values.forEach(it -> bsonArray.add(BsonUtils.simpleToBsonValue(it)));
+					} else if (ObjectUtils.isArray(value)) {
+						for (Object o : ObjectUtils.toObjectArray(value)) {
+							bsonArray.add(BsonUtils.simpleToBsonValue(o));
+						}
+					}
+					return bsonArray;
+				} else {
+					if (value instanceof Collection values) {
+						values.forEach(it -> {
+							Document write = (Document) context.write(it, persistentProperty.getTypeInformation());
+							bsonArray.add(write.toBsonDocument());
+						});
+					} else if (ObjectUtils.isArray(value)) {
+						for (Object o : ObjectUtils.toObjectArray(value)) {
+							Document write = (Document) context.write(0, persistentProperty.getTypeInformation());
+							bsonArray.add(write.toBsonDocument());
+						}
+					}
+					return bsonArray;
+				}
+			}
+
+			if (!persistentProperty.isEntity()) {
+				if (persistentProperty.isCollectionLike()) {
+
+					if (persistentProperty.isEntity()) {
+
+					}
+				}
+			}
+
+			return null;
+		}
+
+		public Object decrypt(Object value, ClientEncryption clientEncryption) {
+
+			// this was a hack to avoid the 60 sec timeout of the key cache
+			// ClientEncryptionSettings settings = (ClientEncryptionSettings) new DirectFieldAccessor(clientEncryption)
+			// .getPropertyValue("options");
+			// clientEncryption = ClientEncryptions.create(settings);
+
+			Object result = value;
+			if (value instanceof Binary binary) {
+				result = clientEncryption.decrypt(new BsonBinary(binary.getType(), binary.getData()));
+			}
+			if (value instanceof BsonBinary binary) {
+				result = clientEncryption.decrypt(binary);
+			}
+
+			// in case the driver has auto decryption (aka .bypassAutoEncryption(true)) active
+			// https://github.com/mongodb/mongo-java-driver/blob/master/driver-sync/src/examples/tour/ClientSideEncryptionExplicitEncryptionOnlyTour.java
+			if (value == result) {
+				return result;
+			}
+
+			if (persistentProperty.isCollectionLike() && result instanceof Iterable<?> iterable) {
+				if (!persistentProperty.isEntity()) {
+					Collection<Object> collection = CollectionFactory.createCollection(persistentProperty.getType(), 10);
+					iterable.forEach(it -> collection.add(BsonUtils.toJavaType((BsonValue) it)));
+					return collection;
+				} else {
+					Collection<Object> collection = CollectionFactory.createCollection(persistentProperty.getType(), 10);
+					iterable.forEach(it -> {
+						collection.add(context.read(BsonUtils.toJavaType((BsonValue) it), persistentProperty.getActualType()));
+					});
+					return collection;
+				}
+			}
+
+			if (!persistentProperty.isEntity() && result instanceof BsonValue bsonValue) {
+				return BsonUtils.toJavaType(bsonValue);
+			}
+
+			if (persistentProperty.isEntity() && result instanceof BsonDocument bsonDocument) {
+				return context.read(BsonUtils.toJavaType(bsonDocument), persistentProperty.getTypeInformation());
+			}
+
+			return result;
+		}
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/fle/ReactiveFLETests.java

@@ -0,0 +1,440 @@
+/*
+ * Copyright 2022 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.mongodb.fle;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.mongodb.core.EncryptionAlgorithms.*;
+import static org.springframework.data.mongodb.core.query.Criteria.*;
+
+import lombok.Data;
+import org.springframework.data.mongodb.fle.FLETests.Person;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.security.SecureRandom;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
+
+import org.bson.BsonArray;
+import org.bson.BsonBinary;
+import org.bson.BsonDocument;
+import org.bson.BsonValue;
+import org.bson.Document;
+import org.bson.types.Binary;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.reactivestreams.Publisher;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.CollectionFactory;
+import org.springframework.core.annotation.AliasFor;
+import org.springframework.data.convert.PropertyValueConverterFactory;
+import org.springframework.data.convert.ValueConverter;
+import org.springframework.data.mongodb.config.AbstractReactiveMongoConfiguration;
+import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
+import org.springframework.data.mongodb.core.convert.MongoConversionContext;
+import org.springframework.data.mongodb.core.convert.MongoCustomConversions.MongoConverterConfigurationAdapter;
+import org.springframework.data.mongodb.core.convert.MongoValueConverter;
+import org.springframework.data.mongodb.core.mapping.Encrypted;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
+import org.springframework.data.mongodb.fle.FLETests.Config;
+import org.springframework.data.mongodb.util.BsonUtils;
+import org.springframework.data.util.Lazy;
+import org.springframework.lang.Nullable;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
+
+import com.mongodb.ClientEncryptionSettings;
+import com.mongodb.ConnectionString;
+import com.mongodb.MongoClientSettings;
+import com.mongodb.MongoNamespace;
+import com.mongodb.client.model.Filters;
+import com.mongodb.client.model.IndexOptions;
+import com.mongodb.client.model.Indexes;
+import com.mongodb.client.model.vault.DataKeyOptions;
+import com.mongodb.client.model.vault.EncryptOptions;
+import com.mongodb.reactivestreams.client.MongoClient;
+import com.mongodb.reactivestreams.client.MongoCollection;
+import com.mongodb.reactivestreams.client.vault.ClientEncryption;
+import com.mongodb.reactivestreams.client.vault.ClientEncryptions;
+
+/**
+ * @author Christoph Strobl
+ * @since 2022/11
+ */
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration(classes = ReactiveFLETests.Config.class)
+public class ReactiveFLETests {
+
+	ClientEncryption encryption;
+
+	@Test
+	void xxx() {
+
+		Supplier<String> valueSupplier = new Supplier<String>() {
+			@Override
+			public String get() {
+				System.out.println("invoked");
+				return "v1";
+			}
+		};
+
+		Document source = new Document("name", "value").append("mono", Mono.fromSupplier(() -> "from mono"))
+				.append("nested", new Document("n1", Mono.fromSupplier(() -> "from nested mono")));
+
+		resolveValues(Mono.just(source)) //
+				.as(StepVerifier::create).consumeNextWith(resolved -> {
+					assertThat(resolved).isEqualTo(Document
+							.parse("{\"name\": \"value\", \"mono\": \"from mono\", \"nested\" : { \"n1\" : \"from nested mono\"}}"));
+				}).verifyComplete();
+	}
+
+	private Mono<Document> resolveValues(Mono<Document> document) {
+		return document.flatMap(source -> {
+			for (Entry<String, Object> entry : source.entrySet()) {
+				if (entry.getValue()instanceof Mono<?> valueMono) {
+					return valueMono.flatMap(value -> {
+						source.put(entry.getKey(), value);
+						return resolveValues(Mono.just(source));
+					});
+				}
+				if (entry.getValue()instanceof Document nested) {
+					return resolveValues(Mono.just(nested)).map(it -> {
+						source.put(entry.getKey(), it);
+						return source;
+					});
+				}
+			}
+			return Mono.just(source);
+		});
+	}
+
+	@Autowired ReactiveMongoTemplate template;
+
+	@Test
+	void manualEnAndDecryption() {
+
+		Person person = new Person();
+		person.id = "id-1";
+		person.name = "p1-name";
+		person.ssn = "mySecretSSN";
+
+		template.save(person).block();
+
+		System.out.println("source: " + person);
+
+		Flux<Document> result = template.execute(FLETests.Person.class, collection -> {
+			return Mono.from(collection.find(new Document()).first());
+		});
+
+		System.out.println("encrypted: " + result.blockFirst().toJson());
+
+		Person id = template.query(Person.class).matching(where("id").is(person.id)).first().block();
+		System.out.println("decrypted: " + id);
+	}
+
+	@Data
+	@org.springframework.data.mongodb.core.mapping.Document("test")
+	static class Person {
+
+		String id;
+		String name;
+
+		@EncryptedField(algorithm = AEAD_AES_256_CBC_HMAC_SHA_512_Deterministic) //
+		String ssn;
+	}
+
+	@Configuration
+	static class Config extends AbstractReactiveMongoConfiguration {
+
+		@Autowired ApplicationContext applicationContext;
+
+		@Override
+		protected String getDatabaseName() {
+			return "fle-test";
+		}
+
+		@Override
+		protected void configureConverters(MongoConverterConfigurationAdapter converterConfigurationAdapter) {
+
+			converterConfigurationAdapter
+					.registerPropertyValueConverterFactory(PropertyValueConverterFactory.beanFactoryAware(applicationContext));
+		}
+
+		@Bean
+		@Override
+		public MongoClient reactiveMongoClient() {
+			return super.reactiveMongoClient();
+		}
+
+		@Bean
+		ReactiveEncryptingConverter encryptingConverter(ClientEncryption clientEncryption) {
+			return new ReactiveEncryptingConverter(clientEncryption);
+		}
+
+		@Bean
+		ClientEncryption clientEncryption(MongoClient mongoClient) {
+
+			final byte[] localMasterKey = new byte[96];
+			new SecureRandom().nextBytes(localMasterKey);
+			Map<String, Map<String, Object>> kmsProviders = new HashMap<String, Map<String, Object>>() {
+				{
+					put("local", new HashMap<String, Object>() {
+						{
+							put("key", localMasterKey);
+						}
+					});
+				}
+			};
+
+			MongoNamespace keyVaultNamespace = new MongoNamespace("encryption.testKeyVault");
+			MongoCollection<Document> keyVaultCollection = mongoClient.getDatabase(keyVaultNamespace.getDatabaseName())
+					.getCollection(keyVaultNamespace.getCollectionName());
+			Mono.from(keyVaultCollection.drop()).block();
+			// Ensure that two data keys cannot share the same keyAltName.
+			keyVaultCollection.createIndex(Indexes.ascending("keyAltNames"),
+					new IndexOptions().unique(true).partialFilterExpression(Filters.exists("keyAltNames")));
+
+			MongoCollection<Document> collection = mongoClient.getDatabase(getDatabaseName()).getCollection("test");
+			Mono.from(collection.drop()).block(); // Clear old data
+
+			// Create the ClientEncryption instance
+			ClientEncryptionSettings clientEncryptionSettings = ClientEncryptionSettings.builder()
+					.keyVaultMongoClientSettings(
+							MongoClientSettings.builder().applyConnectionString(new ConnectionString("mongodb://localhost")).build())
+					.keyVaultNamespace(keyVaultNamespace.getFullName()).kmsProviders(kmsProviders).build();
+			ClientEncryption clientEncryption = ClientEncryptions.create(clientEncryptionSettings);
+			return clientEncryption;
+		}
+	}
+
+	@Retention(RetentionPolicy.RUNTIME)
+	@Target(ElementType.FIELD)
+	@Encrypted
+	@ValueConverter(ReactiveEncryptingConverter.class)
+	@interface EncryptedField {
+
+		@AliasFor(annotation = Encrypted.class, value = "algorithm")
+		String algorithm() default "";
+
+		String altKeyName() default "";
+	}
+
+	static class ReactiveEncryptingConverter implements MongoValueConverter<Object, Object> {
+
+		private ClientEncryption clientEncryption;
+		private BsonBinary dataKeyId; // should be provided from outside.
+
+		public ReactiveEncryptingConverter(ClientEncryption clientEncryption) {
+
+			this.clientEncryption = clientEncryption;
+			this.dataKeyId = Mono.from(clientEncryption.createDataKey("local",
+					new DataKeyOptions().keyAltNames(Collections.singletonList("mySuperSecretKey")))).block();
+		}
+
+		@Nullable
+		@Override
+		public Object read(Object value, MongoConversionContext context) {
+
+			ManualEncryptionContext encryptionContext = buildEncryptionContext(context);
+			Object decrypted = null;
+			try {
+				decrypted = encryptionContext.decrypt(value, clientEncryption);
+			} catch (ExecutionException e) {
+				throw new RuntimeException(e);
+			} catch (InterruptedException e) {
+				throw new RuntimeException(e);
+			}
+			return decrypted instanceof BsonValue ? BsonUtils.toJavaType((BsonValue) decrypted) : decrypted;
+		}
+
+		@Nullable
+		@Override
+		public Publisher<BsonBinary> write(Object value, MongoConversionContext context) {
+
+			ManualEncryptionContext encryptionContext = buildEncryptionContext(context);
+			return encryptionContext.encrypt(value, clientEncryption);
+		}
+
+		ManualEncryptionContext buildEncryptionContext(MongoConversionContext context) {
+			return new ManualEncryptionContext(context, this.dataKeyId);
+		}
+	}
+
+	static class ManualEncryptionContext {
+
+		MongoConversionContext context;
+		MongoPersistentProperty persistentProperty;
+		BsonBinary dataKeyId;
+		Lazy<Encrypted> encryption;
+
+		public ManualEncryptionContext(MongoConversionContext context, BsonBinary dataKeyId) {
+			this.context = context;
+			this.persistentProperty = context.getProperty();
+			this.dataKeyId = dataKeyId;
+			this.encryption = Lazy.of(() -> persistentProperty.findAnnotation(Encrypted.class));
+		}
+
+		Publisher<BsonBinary> encrypt(Object value, ClientEncryption clientEncryption) {
+
+			// TODO: check - encryption.get().keyId()
+
+			EncryptOptions encryptOptions = new EncryptOptions(encryption.get().algorithm());
+
+			EncryptedField annotation = persistentProperty.findAnnotation(EncryptedField.class);
+			if (annotation != null && !annotation.altKeyName().isBlank()) {
+				if (annotation.altKeyName().startsWith("/")) {
+					String fieldName = annotation.altKeyName().replace("/", "");
+					Object altKeyNameValue = context.getValue(fieldName);
+					encryptOptions = encryptOptions.keyAltName(altKeyNameValue.toString());
+				} else {
+					encryptOptions = encryptOptions.keyAltName(annotation.altKeyName());
+				}
+			} else {
+				encryptOptions = encryptOptions.keyId(this.dataKeyId);
+			}
+
+			System.out.println(
+					"encrypting with: " + (StringUtils.hasText(encryptOptions.getKeyAltName()) ? encryptOptions.getKeyAltName()
+							: encryptOptions.getKeyId()));
+
+			if (!persistentProperty.isEntity()) {
+
+				if (persistentProperty.isCollectionLike()) {
+					return clientEncryption.encrypt(collectionLikeToBsonValue(value), encryptOptions);
+				}
+				return clientEncryption.encrypt(BsonUtils.simpleToBsonValue(value), encryptOptions);
+			}
+			if (persistentProperty.isCollectionLike()) {
+				return clientEncryption.encrypt(collectionLikeToBsonValue(value), encryptOptions);
+			}
+
+			Object write = context.write(value);
+			if (write instanceof Document doc) {
+				return clientEncryption.encrypt(doc.toBsonDocument(), encryptOptions);
+			}
+			return clientEncryption.encrypt(BsonUtils.simpleToBsonValue(write), encryptOptions);
+		}
+
+		public BsonValue collectionLikeToBsonValue(Object value) {
+
+			if (persistentProperty.isCollectionLike()) {
+
+				BsonArray bsonArray = new BsonArray();
+				if (!persistentProperty.isEntity()) {
+					if (value instanceof Collection values) {
+						values.forEach(it -> bsonArray.add(BsonUtils.simpleToBsonValue(it)));
+					} else if (ObjectUtils.isArray(value)) {
+						for (Object o : ObjectUtils.toObjectArray(value)) {
+							bsonArray.add(BsonUtils.simpleToBsonValue(o));
+						}
+					}
+					return bsonArray;
+				} else {
+					if (value instanceof Collection values) {
+						values.forEach(it -> {
+							Document write = (Document) context.write(it, persistentProperty.getTypeInformation());
+							bsonArray.add(write.toBsonDocument());
+						});
+					} else if (ObjectUtils.isArray(value)) {
+						for (Object o : ObjectUtils.toObjectArray(value)) {
+							Document write = (Document) context.write(o, persistentProperty.getTypeInformation());
+							bsonArray.add(write.toBsonDocument());
+						}
+					}
+					return bsonArray;
+				}
+			}
+
+			if (!persistentProperty.isEntity()) {
+				if (persistentProperty.isCollectionLike()) {
+
+					if (persistentProperty.isEntity()) {
+
+					}
+				}
+			}
+
+			return null;
+		}
+
+		public Object decrypt(Object value, ClientEncryption clientEncryption) throws ExecutionException, InterruptedException {
+
+			// this was a hack to avoid the 60 sec timeout of the key cache
+			// ClientEncryptionSettings settings = (ClientEncryptionSettings) new DirectFieldAccessor(clientEncryption)
+			// .getPropertyValue("options");
+			// clientEncryption = ClientEncryptions.create(settings);
+
+			Object r = value;
+			if (value instanceof Binary binary) {
+				r = clientEncryption.decrypt(new BsonBinary(binary.getType(), binary.getData()));
+			}
+			if (value instanceof BsonBinary binary) {
+				r = clientEncryption.decrypt(binary);
+			}
+
+			// in case the driver has auto decryption (aka .bypassAutoEncryption(true)) active
+			// https://github.com/mongodb/mongo-java-driver/blob/master/driver-sync/src/examples/tour/ClientSideEncryptionExplicitEncryptionOnlyTour.java
+			if (value == r) {
+				return r;
+			}
+
+			if(r instanceof Mono mono) {
+				return mono.map(result -> {
+					if (persistentProperty.isCollectionLike() && result instanceof Iterable<?> iterable) {
+						if (!persistentProperty.isEntity()) {
+							Collection<Object> collection = CollectionFactory.createCollection(persistentProperty.getType(), 10);
+							iterable.forEach(it -> collection.add(BsonUtils.toJavaType((BsonValue) it)));
+							return collection;
+						} else {
+							Collection<Object> collection = CollectionFactory.createCollection(persistentProperty.getType(), 10);
+							iterable.forEach(it -> {
+								collection.add(context.read(BsonUtils.toJavaType((BsonValue) it), persistentProperty.getActualType()));
+							});
+							return collection;
+						}
+					}
+
+					if (!persistentProperty.isEntity() && result instanceof BsonValue bsonValue) {
+						return BsonUtils.toJavaType(bsonValue);
+					}
+
+					if (persistentProperty.isEntity() && result instanceof BsonDocument bsonDocument) {
+						return context.read(BsonUtils.toJavaType(bsonDocument), persistentProperty.getTypeInformation());
+					}
+					return result;
+				}).toFuture().get();
+			}
+
+			return r;
+		}
+	}
+
+}