Fix MongoDb module for the latest Spring Data

The `MappingMongoConverter` now used different path when it
iterates documents form conversion from a DB cursor

* Fix `MongoDbMessageStore.MessageReadingMongoConverter` to override
a `read(TypeInformation<S>, Bson)` method to convert a `MessageWrapper` properly
* Upgrade to the latest MongoDb driver

build.gradle

@@ -85,12 +85,12 @@ ext {
     mailVersion = '2.0.1'
 	micrometerVersion = '1.10.0-SNAPSHOT'
     mockitoVersion = '4.4.0'
-    mongoDriverVersion = '4.5.0'
+    mongoDriverVersion = '4.6.0'
     mysqlVersion = '8.0.28'
     pahoMqttClientVersion = '1.2.5'
     postgresVersion = '42.3.3'
     r2dbch2Version = '0.9.1.RELEASE'
-    reactorVersion = '2020.0.15'
+    reactorVersion = '2020.0.18'
     resilience4jVersion = '1.7.1'
     romeToolsVersion = '1.18.0'
     rsocketVersion = '1.1.1'

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

@@ -43,7 +43,6 @@ import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.serializer.support.SerializingConverter;
 import org.springframework.data.annotation.Id;
-import org.springframework.data.annotation.Transient;
 import org.springframework.data.convert.ReadingConverter;
 import org.springframework.data.convert.WritingConverter;
 import org.springframework.data.domain.Sort;
@@ -63,6 +62,7 @@ import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
+import org.springframework.data.util.TypeInformation;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.message.AdviceMessage;
 import org.springframework.integration.store.AbstractMessageGroupStore;
@@ -495,9 +495,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
 		Query query = Query.query(Criteria.where("_id").is(SEQUENCE_NAME));
 		query.fields().include(SEQUENCE);
 		return ((Number) this.template.findAndModify(query,
-				new Update().inc(SEQUENCE, 1L),
+						new Update().inc(SEQUENCE, 1L),
-				FindAndModifyOptions.options().returnNew(true).upsert(true),
+						FindAndModifyOptions.options().returnNew(true).upsert(true),
-				Map.class, this.collectionName)
+						Map.class, this.collectionName)
 				.get(SEQUENCE))  // NOSONAR - never returns null
 				.longValue();
 	}
@@ -589,9 +589,14 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
 			if (!MessageWrapper.class.equals(clazz)) {
 				return super.read(clazz, source);
 			}
+
+			return (S) readAsMessageWrapper(source);
+		}
+
+		private MessageWrapper readAsMessageWrapper(Bson source) {
 			if (source != null) {
 				Map<String, Object> sourceMap = asMap(source);
-				Message<?> message = null;
+				Message<?> message;
 				Object messageType = sourceMap.get("_messageType");
 				if (messageType == null) {
 					messageType = GenericMessage.class.getName();
@@ -629,21 +634,30 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
 				}
 				wrapper.setCondition((String) sourceMap.get("_condition"));
 
-				return (S) wrapper;
+				return wrapper;
 			}
 			return null;
 		}
 
+		@Override
+		@SuppressWarnings({ UNCHECKED })
+		protected <S> S read(TypeInformation<S> type, Bson source) {
+			if (!MessageWrapper.class.equals(type.getType())) {
+				return super.read(type, source);
+			}
+
+			return (S) readAsMessageWrapper(source);
+		}
+
 		private Map<String, Object> normalizeHeaders(Map<String, Object> headers) {
 			Map<String, Object> normalizedHeaders = new HashMap<>();
 			for (Entry<String, Object> entry : headers.entrySet()) {
 				String headerName = entry.getKey();
 				Object headerValue = entry.getValue();
-				if (headerValue instanceof Bson) {
+				if (headerValue instanceof Bson source) {
-					Bson source = (Bson) headerValue;
 					Map<String, Object> document = asMap(source);
 					try {
-						Class<?> typeClass = null;
+						Class<?> typeClass;
 						if (document.containsKey(CLASS)) {
 							Object type = document.get(CLASS);
 							typeClass = ClassUtils.forName(type.toString(), MongoDbMessageStore.this.classLoader);
@@ -670,8 +684,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
 		private Object extractPayload(Bson source) {
 			Object payload = asMap(source).get("payload");
 
-			if (payload instanceof Bson) {
+			if (payload instanceof Bson payloadObject) {
-				Bson payloadObject = (Bson) payload;
 				Object payloadType = asMap(payloadObject).get(CLASS);
 				try {
 					Class<?> payloadClass =
@@ -845,7 +858,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
 
 		private volatile Object _groupId; // NOSONAR name
 
-		@Transient
+		//		@Transient
 		private final Message<?> message; // NOSONAR name
 
 		@SuppressWarnings(UNUSED)