INT-4570: Add MessageCollectionCallback for Mongo (#2675)

* INT-4570: Add MessageCollectionCallback for Mongo

JIRA: https://jira.spring.io/browse/INT-4570

The `MongoDbOutboundGateway` is intended to be used with the
`requestMessage` context, however using a plain `CollectionCallback`
we don't have access to the `requestMessage`

* Deprecate `CollectionCallback` usage in favor of newly introduced
`MessageCollectionCallback` and `message-collection-callback` for XML

**Cherry-pick to 5.0.x**

* * Remove `message-collection-callback` in favor of
`MessageCollectionCallback<T> extends CollectionCallback<T>`

* * Rename a new setter to `setMessageCollectionCallback()` to avoid
reflection collision

Author: artembilan

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbOutboundGatewaySpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -26,13 +26,16 @@
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.dsl.MessageHandlerSpec;
 import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.mongodb.outbound.MessageCollectionCallback;
 import org.springframework.integration.mongodb.outbound.MongoDbOutboundGateway;
 import org.springframework.messaging.Message;
 
 /**
  * A {@link MessageHandlerSpec} extension for the MongoDb Outbound endpoint {@link MongoDbOutboundGateway}
  *
  * @author Xavier Padró
+ * @author Artem Bilan
+ *
  * @since 5.0
  */
 public class MongoDbOutboundGatewaySpec
@@ -149,10 +152,26 @@ public <P> MongoDbOutboundGatewaySpec collectionNameFunction(Function<Message<P>
 	 * @param collectionCallback the {@link CollectionCallback} instance
 	 * @param <P> the type of the message payload.
 	 * @return the spec
+	 * @deprecated in favor of {@link #collectionCallback(MessageCollectionCallback)}
 	 */
+	@Deprecated
 	public <P> MongoDbOutboundGatewaySpec collectionCallback(CollectionCallback<P> collectionCallback) {
 		this.target.setCollectionCallback(collectionCallback);
 		return this;
 	}
 
+	/**
+	 * Reference to an instance of {@link MessageCollectionCallback}
+	 * which specifies the database operation to execute in the request message context.
+	 * This property is mutually exclusive with {@link #query} and {@link #queryExpression} properties.
+	 * @param collectionCallback the {@link MessageCollectionCallback} instance
+	 * @param <P> the type of the message payload.
+	 * @return the spec
+	 * @since 5.0.11
+	 */
+	public <P> MongoDbOutboundGatewaySpec collectionCallback(MessageCollectionCallback<P> collectionCallback) {
+		this.target.setMessageCollectionCallback(collectionCallback);
+		return this;
+	}
+
 }

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/outbound/MessageCollectionCallback.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2018 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.integration.mongodb.outbound;
+
+import org.bson.Document;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.mongodb.core.CollectionCallback;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+
+import com.mongodb.MongoException;
+import com.mongodb.client.MongoCollection;
+
+/**
+ * The callback to be used with the {@link MongoDbOutboundGateway}
+ * as an alternative to other query options on the gateway.
+ * <p>
+ * Plays the same role as standard {@link CollectionCallback},
+ * but with {@code Message<?> requestMessage} context during {@code handleMessage()}
+ * process in the {@link MongoDbOutboundGateway}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0.11
+ *
+ * @see CollectionCallback
+ */
+@FunctionalInterface
+public interface MessageCollectionCallback<T> extends CollectionCallback<T> {
+
+	/**
+	 * Perform a Mongo operation in the collection using request message as a context.
+	 * @param collection never {@literal null}.
+	 * @param requestMessage the request message ot use for operations
+	 * @return can be {@literal null}.
+	 * @throws MongoException the MongoDB-specific exception
+	 * @throws DataAccessException the data access exception
+	 */
+	@Nullable
+	T doInCollection(MongoCollection<Document> collection, Message<?> requestMessage)
+			throws MongoException, DataAccessException;
+
+	@Override
+	default T doInCollection(MongoCollection<Document> collection) throws MongoException, DataAccessException {
+		throw new UnsupportedOperationException("The 'doInCollection(MongoCollection<Document>, Message<?>)' " +
+				"must be implemented instead.");
+	}
+
+}

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/outbound/MongoDbOutboundGateway.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -41,6 +41,8 @@
  * Makes outbound operations to query a MongoDb database using a {@link MongoOperations}
  *
  * @author Xavier Padró
+ * @author Artem Bilan
+ *
  * @since 5.0
  */
 public class MongoDbOutboundGateway extends AbstractReplyProducingMessageHandler {
@@ -55,7 +57,7 @@ public class MongoDbOutboundGateway extends AbstractReplyProducingMessageHandler
 
 	private Expression queryExpression;
 
-	private CollectionCallback<?> collectionCallback;
+	private MessageCollectionCallback<?> collectionCallback;
 
 	private boolean expectSingleResult = false;
 
@@ -90,8 +92,29 @@ public void setQueryExpressionString(String queryExpressionString) {
 		this.queryExpression = EXPRESSION_PARSER.parseExpression(queryExpressionString);
 	}
 
+	/**
+	 * Specify a {@link CollectionCallback} to perform against MongoDB collection.
+	 * @param collectionCallback the callback to perform against MongoDB collection.
+	 * @deprecated in favor of {@link #setMessageCollectionCallback(MessageCollectionCallback)}.
+	 * Will be removed in 5.2
+	 */
+	@Deprecated
 	public void setCollectionCallback(CollectionCallback<?> collectionCallback) {
-		Assert.notNull(collectionCallback, "collectionCallback must not be null.");
+		Assert.notNull(collectionCallback, "'collectionCallback' must not be null.");
+		this.collectionCallback =
+				collectionCallback instanceof MessageCollectionCallback
+						? (MessageCollectionCallback) collectionCallback
+						: (collection, requestMessage) -> collectionCallback.doInCollection(collection);
+	}
+
+	/**
+	 * Specify a {@link MessageCollectionCallback} to perform against MongoDB collection
+	 * in the request message context.
+	 * @param collectionCallback the callback to perform against MongoDB collection.
+	 * @since 5.0.11
+	 */
+	public void setMessageCollectionCallback(MessageCollectionCallback<?> collectionCallback) {
+		Assert.notNull(collectionCallback, "'collectionCallback' must not be null.");
 		this.collectionCallback = collectionCallback;
 	}
 
@@ -152,7 +175,8 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 		Object result;
 
 		if (this.collectionCallback != null) {
-			result = this.mongoTemplate.execute(collectionName, this.collectionCallback); // NOSONAR
+			result = this.mongoTemplate.execute(collectionName, // NOSONAR
+					collection -> this.collectionCallback.doInCollection(collection, requestMessage));
 		}
 		else {
 			Query query = buildQuery(requestMessage);

spring-integration-mongodb/src/main/resources/org/springframework/integration/mongodb/config/spring-integration-mongodb-5.1.xsd

@@ -244,7 +244,10 @@
 							<xsd:appinfo>
 								<xsd:documentation>
 									Reference to an instance of
-									org.springframework.data.mongodb.core.CollectionCallback
+									org.springframework.data.mongodb.core.CollectionCallback, preferable an
+									instance of
+									org.springframework.integration.mongodb.outbound.MessageCollectionCallback
+									with the request message context.
 								</xsd:documentation>
 								<tool:annotation kind="ref">
 									<tool:expected-type

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbOutboundGatewayParserTests.java

@@ -34,13 +34,13 @@
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.data.mongodb.MongoDbFactory;
-import org.springframework.data.mongodb.core.CollectionCallback;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.standard.SpelExpression;
 import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
+import org.springframework.integration.mongodb.outbound.MessageCollectionCallback;
 import org.springframework.integration.mongodb.outbound.MongoDbOutboundGateway;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.MessageHandler;
@@ -143,7 +143,7 @@ public void fullConfigWithMongoDbCollectionCallback() {
 				instanceOf(LiteralExpression.class));
 		assertEquals("foo", TestUtils.getPropertyValue(gateway, "collectionNameExpression.literalValue"));
 		assertThat(TestUtils.getPropertyValue(gateway, "collectionCallback"),
-				instanceOf(CollectionCallback.class));
+				instanceOf(MessageCollectionCallback.class));
 	}
 
 	@Test(expected = BeanDefinitionParsingException.class)

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/dsl/MongoDbTests.java

@@ -33,7 +33,6 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.core.BulkOperations;
-import org.springframework.data.mongodb.core.CollectionCallback;
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
@@ -45,6 +44,7 @@
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.handler.ReplyRequiredException;
+import org.springframework.integration.mongodb.outbound.MessageCollectionCallback;
 import org.springframework.integration.mongodb.rules.MongoDbAvailable;
 import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
 import org.springframework.integration.support.MessageBuilder;
@@ -55,11 +55,11 @@
 import org.springframework.test.context.junit4.SpringRunner;
 
 import com.mongodb.MongoClient;
-import com.mongodb.client.MongoCollection;
 
 /**
  * @author Xavier Padró
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.0
  */
@@ -315,7 +315,8 @@ public IntegrationFlow gatewayCollectionNameFunctionFlow() {
 		@Bean
 		public IntegrationFlow gatewayCollectionCallbackFlow() {
 			return f -> f
-					.handle(collectionCallbackOutboundGateway(MongoCollection::countDocuments))
+					.handle(collectionCallbackOutboundGateway(
+							(collection, requestMessage) -> collection.countDocuments()))
 					.channel(getResultChannel());
 		}
 
@@ -389,7 +390,9 @@ private MongoDbOutboundGatewaySpec collectionNameFunctionOutboundGateway(boolean
 					.entityClass(Person.class);
 		}
 
-		private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway(CollectionCallback<?> collectionCallback) {
+		private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway(
+				MessageCollectionCallback<?> collectionCallback) {
+
 			return MongoDb.outboundGateway(mongoDbFactory(), mongoConverter())
 					.collectionCallback(collectionCallback)
 					.collectionName(COLLECTION_NAME)