GH-2936: Add custom converters for MongoDbMS

Fixes #2936

* Add `MongoDbMessageStore.setCustomConverters()` to allow to inject
any custom converters for `payload` as well as header values

**Cherry-pick to 5.1.x**

Author: artembilan

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java

@@ -17,7 +17,9 @@
 package org.springframework.integration.mongodb.store;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -101,14 +103,14 @@
 public class MongoDbMessageStore extends AbstractMessageGroupStore
 		implements MessageStore, BeanClassLoaderAware, ApplicationContextAware, InitializingBean {
 
+	public static final String SEQUENCE_NAME = "messagesSequence";
+
 	private static final String HEADERS = "headers";
 
 	private static final String UNCHECKED = "unchecked";
 
 	private static final String GROUP_ID_MUST_NOT_BE_NULL = "'groupId' must not be null";
 
-	public static final String SEQUENCE_NAME = "messagesSequence";
-
 	private static final String DEFAULT_COLLECTION_NAME = "messages";
 
 	private static final String GROUP_ID_KEY = "_groupId";
@@ -132,13 +134,12 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
 
 	private final String collectionName;
 
-	private volatile ClassLoader classLoader = ClassUtils.getDefaultClassLoader();
+	private ClassLoader classLoader = ClassUtils.getDefaultClassLoader();
 
 	private ApplicationContext applicationContext;
 
 	private String[] whiteListPatterns;
 
-
 	/**
 	 * Create a MongoDbMessageStore using the provided {@link MongoDbFactory}.and the default collection name.
 	 * @param mongoDbFactory The mongodb factory.
@@ -178,14 +179,28 @@ public void setApplicationContext(ApplicationContext applicationContext) throws
 	 * @param patterns the patterns.
 	 */
 	public void addWhiteListPatterns(String... patterns) {
-		this.whiteListPatterns = patterns;
+		this.whiteListPatterns = patterns != null ? Arrays.copyOf(patterns, patterns.length) : null;
+	}
+
+	/**
+	 * Configure a set of converters to use in the {@link MappingMongoConverter}.
+	 * Must be instances of {@code org.springframework.core.convert.converter.Converter},
+	 * {@code org.springframework.core.convert.converter.ConverterFactory},
+	 * {@code org.springframework.core.convert.converter.GenericConverter} or
+	 * {@code org.springframework.data.convert.ConverterBuilder.ConverterAware}.
+	 * @param customConverters the converters to use.
+	 * @since 5.1.6
+	 */
+	public void setCustomConverters(Object... customConverters) {
+		this.converter.setCustomConverters(customConverters);
 	}
 
 	@Override
 	public void afterPropertiesSet() {
 		if (this.applicationContext != null) {
 			this.converter.setApplicationContext(this.applicationContext);
 		}
+
 		this.converter.afterPropertiesSet();
 
 		IndexOperations indexOperations = this.template.indexOps(this.collectionName);
@@ -506,26 +521,38 @@ private final class MessageReadingMongoConverter extends MappingMongoConverter {
 
 		private static final String CLASS = "_class";
 
+		private Object[] customConverters;
+
 		MessageReadingMongoConverter(MongoDbFactory mongoDbFactory,
 				MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext) {
 			super(new DefaultDbRefResolver(mongoDbFactory), mappingContext);
 		}
 
+		void setCustomConverters(Object... customConverters) {
+			this.customConverters =
+					customConverters != null ? Arrays.copyOf(customConverters, customConverters.length) : null;
+		}
+
 		@Override
 		public void afterPropertiesSet() {
-			List<Object> customConverters = new ArrayList<>();
-			customConverters.add(new MessageHistoryToDocumentConverter());
-			customConverters.add(new DocumentToGenericMessageConverter());
-			customConverters.add(new DocumentToMutableMessageConverter());
+			List<Object> converters = new ArrayList<>();
+			converters.add(new MessageHistoryToDocumentConverter());
+			converters.add(new DocumentToGenericMessageConverter());
+			converters.add(new DocumentToMutableMessageConverter());
 			DocumentToErrorMessageConverter docToErrorMessageConverter = new DocumentToErrorMessageConverter();
 			if (MongoDbMessageStore.this.whiteListPatterns != null) {
 				docToErrorMessageConverter.deserializingConverter
 						.addWhiteListPatterns(MongoDbMessageStore.this.whiteListPatterns);
 			}
-			customConverters.add(docToErrorMessageConverter);
-			customConverters.add(new DocumentToAdviceMessageConverter());
-			customConverters.add(new ThrowableToBytesConverter());
-			this.setCustomConversions(new MongoCustomConversions(customConverters));
+			converters.add(docToErrorMessageConverter);
+			converters.add(new DocumentToAdviceMessageConverter());
+			converters.add(new ThrowableToBytesConverter());
+
+			if (this.customConverters != null) {
+				Collections.addAll(converters, this.customConverters);
+			}
+
+			setCustomConversions(new MongoCustomConversions(converters));
 			super.afterPropertiesSet();
 		}
 

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreTests.java

@@ -16,8 +16,18 @@
 
 package org.springframework.integration.mongodb.store;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.WritingConverter;
 import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
 import org.springframework.integration.store.MessageStore;
+import org.springframework.messaging.support.GenericMessage;
 
 import com.mongodb.MongoClient;
 
@@ -30,11 +40,56 @@
 public class MongoDbMessageStoreTests extends AbstractMongoDbMessageStoreTests {
 
 	@Override
-	protected MessageStore getMessageStore() throws Exception {
+	protected MessageStore getMessageStore() {
 		MongoDbMessageStore mongoDbMessageStore =
 				new MongoDbMessageStore(new SimpleMongoDbFactory(new MongoClient(), "test"));
 		mongoDbMessageStore.afterPropertiesSet();
 		return mongoDbMessageStore;
 	}
 
+	@Test
+	public void testCustomConverter() throws InterruptedException {
+		MongoDbMessageStore mongoDbMessageStore =
+				new MongoDbMessageStore(new SimpleMongoDbFactory(new MongoClient(), "test"));
+		FooToBytesConverter fooToBytesConverter = new FooToBytesConverter();
+		mongoDbMessageStore.setCustomConverters(fooToBytesConverter);
+		mongoDbMessageStore.afterPropertiesSet();
+
+		mongoDbMessageStore.addMessage(new GenericMessage<>(new Foo("foo")));
+
+		assertThat(fooToBytesConverter.called.await(10, TimeUnit.SECONDS)).isTrue();
+	}
+
+	private static class Foo {
+
+		String foo;
+
+		Foo(String foo) {
+			this.foo = foo;
+		}
+
+		@Override
+		public String toString() {
+			return foo;
+		}
+
+	}
+
+	@WritingConverter
+	private static class FooToBytesConverter implements Converter<Foo, byte[]> {
+
+		private CountDownLatch called = new CountDownLatch(1);
+
+		@Override
+		public byte[] convert(Foo source) {
+			try {
+				return source.toString().getBytes();
+			}
+			finally {
+				this.called.countDown();
+			}
+		}
+
+	}
+
 }

src/reference/asciidoc/mongodb.adoc

@@ -123,9 +123,9 @@ The `MongoDbMessageStore` expands the `Message` as a Mongo document with all nes
 It is useful when you need to have access to the `payload` or `headers` for auditing or analytics -- for example, against stored messages.
 
 IMPORTANT: The `MongoDbMessageStore` uses a custom `MappingMongoConverter` implementation to store `Message` instances as MongoDB documents, and there are some limitations for the properties (`payload` and `header` values) of the `Message`.
-For example, there is no ability to configure custom converters for complex domain `payload` instances or `header` values.
-There is also no way to provide a custom `MongoTemplate` (or `MappingMongoConverter`).
-To achieve these capabilities, an alternative MongoDB `MessageStore` implementation has been introduced (we describe it next).
+
+Starting with version 5.1.6, the `MongoDbMessageStore` can be configured with custom converters which are propagated into an internal `MappingMongoConverter` implementation.
+See `MongoDbMessageStore.setCustomConverters(Object... customConverters)` JavaDocs for more information.
 
 Spring Integration 3.0 introduced the `ConfigurableMongoDbMessageStore`.
 It implements both the `MessageStore` and `MessageGroupStore` interfaces.

src/reference/asciidoc/whats-new.adoc

@@ -69,3 +69,9 @@ See <<mail-inbound>> for more information.
 
 The `WebFluxRequestExecutingMessageHandler` now supports a `Publisher`, `Resource` and `MultiValueMap` as a request message `payload`.
 See <<webflux>> for more information.
+
+[[x5.2-mongodb]]
+==== MongoDb Changes
+
+The `MongoDbMessageStore` can now be configured with custom converters.
+See <<mongodb>> for more information.