INT-4196: MongoOutGateway: add CollectionCallback

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

Add `CollectionCallback` option to the `MongoDbOutboundGateway`

Add XML support to CollectionCallback

Fix PR comments

Fix issue with javadoc parsing

* Polishing code style, JavaDocs and some Docs

spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java

@@ -950,7 +950,7 @@ public class EnableIntegrationTests {
 		@Bean
 		@ServiceActivator(inputChannel = "controlBusChannel")
 		@Role("bar")
-		public ExpressionControlBusFactoryBean controlBus() throws Exception {
+		public ExpressionControlBusFactoryBean controlBus() {
 			return new ExpressionControlBusFactoryBean();
 		}
 

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbOutboundGatewayParser.java

@@ -42,18 +42,28 @@ public class MongoDbOutboundGatewayParser extends AbstractConsumerEndpointParser
 		MongoParserUtils.processCommonAttributes(element, parserContext, builder);
 
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout");
-		String replyChannel = element.getAttribute("reply-channel");
 
-		if (StringUtils.hasText(replyChannel)) {
-			builder.addPropertyReference("outputChannel", replyChannel);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
+		String collectionCallback = element.getAttribute("collection-callback");
+
+		if (StringUtils.hasText(collectionCallback)) {
+			if (StringUtils.hasText(element.getAttribute("query")) ||
+							StringUtils.hasText(element.getAttribute("query-expression"))) {
+
+				parserContext.getReaderContext().error("'collection-callback' is not allowed with " +
+						"'query' or 'query-expression'", element);
+			}
+
+			builder.addPropertyReference("collectionCallback", collectionCallback);
 		}
+		else {
+			BeanDefinition queryExpressionDef =
+					IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("query",
+							"query-expression", parserContext, element, true);
 
-		BeanDefinition queryExpressionDef =
-				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("query",
-						"query-expression", parserContext, element, true);
-
-		if (queryExpressionDef != null) {
-			builder.addPropertyValue("queryExpression", queryExpressionDef);
+			if (queryExpressionDef != null) {
+				builder.addPropertyValue("queryExpression", queryExpressionDef);
+			}
 		}
 
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-single-result");

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

@@ -28,12 +28,25 @@ import org.springframework.data.mongodb.core.convert.MongoConverter;
  */
 public final class MongoDb {
 
+	/**
+	 * Create a {@link MongoDbOutboundGatewaySpec} builder instance
+	 * based on the provided {@link MongoDbFactory} and {@link MongoConverter}.
+	 * @param mongoDbFactory the {@link MongoDbFactory} to use.
+	 * @param mongoConverter the {@link MongoConverter} to use.
+	 * @return the {@link MongoDbOutboundGatewaySpec} instance
+	 */
 	public static MongoDbOutboundGatewaySpec outboundGateway(
 			MongoDbFactory mongoDbFactory, MongoConverter mongoConverter) {
 
 		return new MongoDbOutboundGatewaySpec(mongoDbFactory, mongoConverter);
 	}
 
+	/**
+	 * Create a {@link MongoDbOutboundGatewaySpec} builder instance
+	 * based on the provided {@link MongoOperations}.
+	 * @param mongoTemplate the {@link MongoOperations} to use.
+	 * @return the {@link MongoDbOutboundGatewaySpec} instance
+	 */
 	public static MongoDbOutboundGatewaySpec outboundGateway(MongoOperations mongoTemplate) {
 		return new MongoDbOutboundGatewaySpec(mongoTemplate);
 	}

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

@@ -19,6 +19,7 @@ package org.springframework.integration.mongodb.dsl;
 import java.util.function.Function;
 
 import org.springframework.data.mongodb.MongoDbFactory;
+import org.springframework.data.mongodb.core.CollectionCallback;
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.query.Query;
@@ -47,44 +48,111 @@ public class MongoDbOutboundGatewaySpec
 		this.target.setRequiresReply(true);
 	}
 
+	/**
+	 * This parameter indicates that only one result object will be returned from the database
+	 * by using a {@code findOne} query.
+	 * If set to {@code false} (default), the complete result list is returned as the payload.
+	 * @param expectSingleResult the {@code boolean} flag to indicate if a single result is returned or not.
+	 * @return the spec
+	 */
 	public MongoDbOutboundGatewaySpec expectSingleResult(boolean expectSingleResult) {
 		this.target.setExpectSingleResult(expectSingleResult);
 		return this;
 	}
 
+	/**
+	 * A {@code String} representation of a MongoDb {@link Query} (e.g., query("{'name' : 'Bob'}")).
+	 * Please refer to MongoDb documentation for more query samples
+	 * see <a href="http://www.mongodb.org/display/DOCS/Querying">MongoDB Docs</a>
+	 * This property is mutually exclusive with 'queryExpression' property.
+	 * @param query the MongoDb {@link Query} string representation to use.
+	 * @return the spec
+	 */
 	public MongoDbOutboundGatewaySpec query(String query) {
 		this.target.setQueryExpression(new LiteralExpression(query));
 		return this;
 	}
 
+	/**
+	 * A SpEL expression which should resolve to a {@code String} query (please refer to the 'query' property),
+	 * or to an instance of MongoDb {@link Query}
+	 * (e.q., queryExpression("new BasicQuery('{''address.state'' : ''PA''}')")).
+	 * @param queryExpression the SpEL expression query to use.
+	 * @return the spec
+	 */
 	public MongoDbOutboundGatewaySpec queryExpression(String queryExpression) {
 		this.target.setQueryExpressionString(queryExpression);
 		return this;
 	}
 
+	/**
+	 * A {@link Function} which should resolve to a {@link Query} instance.
+	 * @param queryFunction the {@link Function} to use.
+	 * @param <P> the type of the message payload.
+	 * @return the spec
+	 */
 	public <P> MongoDbOutboundGatewaySpec queryFunction(Function<Message<P>, Query> queryFunction) {
 		this.target.setQueryExpression(new FunctionExpression<>(queryFunction));
 		return this;
 	}
 
+	/**
+	 * The fully qualified name of the entity class to be passed
+	 * to {@code find(..)} or {@code findOne(..)} method in {@link MongoOperations}.
+	 * If this attribute is not provided the default value is {@link org.bson.Document}.
+	 * @param entityClass the {@link Class} to use.
+	 * @return the spec
+	 */
 	public MongoDbOutboundGatewaySpec entityClass(Class<?> entityClass) {
 		this.target.setEntityClass(entityClass);
 		return this;
 	}
 
+	/**
+	 * Identify the name of the MongoDb collection to use.
+	 * This attribute is mutually exclusive with {@link #collectionNameExpression} property.
+	 * @param collectionName the {@link String} specifying the MongoDb collection.
+	 * @return the spec
+	 */
 	public MongoDbOutboundGatewaySpec collectionName(String collectionName) {
 		this.target.setCollectionNameExpression(new LiteralExpression(collectionName));
 		return this;
 	}
 
+	/**
+	 * A SpEL expression which should resolve to a {@link String} value
+	 * identifying the name of the MongoDb collection to use.
+	 * This property is mutually exclusive with {@link #collectionName} property.
+	 * @param collectionNameExpression the {@link String} expression to use.
+	 * @return the spec
+	 */
 	public MongoDbOutboundGatewaySpec collectionNameExpression(String collectionNameExpression) {
 		this.target.setCollectionNameExpressionString(collectionNameExpression);
 		return this;
 	}
 
+	/**
+	 * A {@link Function} which should resolve to a {@link String}
+	 * (e.q., {@code collectionNameFunction(Message::getPayload)}).
+	 * @param collectionNameFunction the {@link Function} to use.
+	 * @param <P> the type of the message payload.
+	 * @return the spec
+	 */
 	public <P> MongoDbOutboundGatewaySpec collectionNameFunction(Function<Message<P>, String> collectionNameFunction) {
 		this.target.setCollectionNameExpression(new FunctionExpression<>(collectionNameFunction));
 		return this;
 	}
 
+	/**
+	 * Reference to an instance of {@link CollectionCallback} which specifies the database operation to execute.
+	 * This property is mutually exclusive with {@link #query} and {@link #queryExpression} properties.
+	 * @param collectionCallback the {@link CollectionCallback} instance
+	 * @param <P> the type of the message payload.
+	 * @return the spec
+	 */
+	public <P> MongoDbOutboundGatewaySpec collectionCallback(CollectionCallback<P> collectionCallback) {
+		this.target.setCollectionCallback(collectionCallback);
+		return this;
+	}
+
 }

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

@@ -19,6 +19,7 @@ package org.springframework.integration.mongodb.outbound;
 import org.bson.Document;
 
 import org.springframework.data.mongodb.MongoDbFactory;
+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.convert.DefaultDbRefResolver;
@@ -54,6 +55,8 @@ public class MongoDbOutboundGateway extends AbstractReplyProducingMessageHandler
 
 	private Expression queryExpression;
 
+	private CollectionCallback<?> collectionCallback;
+
 	private boolean expectSingleResult = false;
 
 	private Class<?> entityClass = Document.class;
@@ -87,6 +90,11 @@ public class MongoDbOutboundGateway extends AbstractReplyProducingMessageHandler
 		this.queryExpression = EXPRESSION_PARSER.parseExpression(queryExpressionString);
 	}
 
+	public void setCollectionCallback(CollectionCallback<?> collectionCallback) {
+		Assert.notNull(collectionCallback, "collectionCallback must not be null.");
+		this.collectionCallback = collectionCallback;
+	}
+
 	public void setExpectSingleResult(boolean expectSingleResult) {
 		this.expectSingleResult = expectSingleResult;
 	}
@@ -115,8 +123,12 @@ public class MongoDbOutboundGateway extends AbstractReplyProducingMessageHandler
 
 	@Override
 	protected void doInit() {
-		Assert.state(this.queryExpression != null, "no query specified");
+		Assert.state(this.queryExpression != null || this.collectionCallback != null,
+				"no query or collectionCallback is specified");
 		Assert.state(this.collectionNameExpression != null, "no collection name specified");
+		if (this.queryExpression != null && this.collectionCallback != null) {
+			throw new IllegalStateException("query and collectionCallback are mutually exclusive");
+		}
 
 		if (this.evaluationContext == null) {
 			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
@@ -136,15 +148,21 @@ public class MongoDbOutboundGateway extends AbstractReplyProducingMessageHandler
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		String collectionName =
 				this.collectionNameExpression.getValue(this.evaluationContext, requestMessage, String.class);
-		Query query = buildQuery(requestMessage);
 
 		Object result;
 
-		if (this.expectSingleResult) {
-			result = this.mongoTemplate.findOne(query, this.entityClass, collectionName);
+		if (this.collectionCallback != null) {
+			result = this.mongoTemplate.execute(collectionName, this.collectionCallback);
 		}
 		else {
-			result = this.mongoTemplate.find(query, this.entityClass, collectionName);
+			Query query = buildQuery(requestMessage);
+
+			if (this.expectSingleResult) {
+				result = this.mongoTemplate.findOne(query, this.entityClass, collectionName);
+			}
+			else {
+				result = this.mongoTemplate.find(query, this.entityClass, collectionName);
+			}
 		}
 
 		return result;

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

@@ -234,6 +234,20 @@
 							</xsd:appinfo>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="collection-callback" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<xsd:documentation>
+									Reference to an instance of
+									org.springframework.data.mongodb.core.CollectionCallback
+								</xsd:documentation>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+											type="org.springframework.data.mongodb.core.CollectionCallback" />
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

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

@@ -39,11 +39,24 @@
 			request-channel="in"
 			reply-channel="out"/>
 
+	<int-mongodb:outbound-gateway id="fullConfigWithMongoDbCollectionCallback"
+			mongo-template="mongoDbTemplate"
+			collection-name="foo"
+			collection-callback="mockCollectionCallback"
+			request-channel="in"
+			reply-channel="out"/>
+
+
+	<bean id="mockCollectionCallback" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.data.mongodb.core.CollectionCallback"/>
+	</bean>
+
 	<bean id="mongoDbFactory" class="org.mockito.Mockito" factory-method="mock">
 		<constructor-arg value="org.springframework.data.mongodb.MongoDbFactory"/>
 	</bean>
 
-	<bean id="mongoConverter" class="org.springframework.integration.mongodb.rules.MongoDbAvailableTests.TestMongoConverter">
+	<bean id="mongoConverter"
+		  class="org.springframework.integration.mongodb.rules.MongoDbAvailableTests.TestMongoConverter">
 		<constructor-arg ref="mongoDbFactory"/>
 		<constructor-arg>
 			<bean class="org.springframework.data.mongodb.core.mapping.MongoMappingContext"/>

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

@@ -16,11 +16,12 @@
 
 package org.springframework.integration.mongodb.config;
 
-import static org.junit.Assert.assertEquals;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -30,6 +31,7 @@ import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
 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;
@@ -65,7 +67,8 @@ public class MongoDbOutboundGatewayParserTests {
 		assertNotNull(TestUtils.getPropertyValue(gateway, "mongoTemplate"));
 		assertSame(this.mongoDbFactory, TestUtils.getPropertyValue(gateway, "mongoDbFactory"));
 		assertNotNull(TestUtils.getPropertyValue(gateway, "evaluationContext"));
-		assertTrue(TestUtils.getPropertyValue(gateway, "collectionNameExpression") instanceof LiteralExpression);
+		assertThat(TestUtils.getPropertyValue(gateway, "collectionNameExpression"),
+				instanceOf(LiteralExpression.class));
 		assertEquals("foo", TestUtils.getPropertyValue(gateway, "collectionNameExpression.literalValue"));
 	}
 
@@ -78,7 +81,8 @@ public class MongoDbOutboundGatewayParserTests {
 		assertSame(this.mongoDbFactory, TestUtils.getPropertyValue(gateway, "mongoDbFactory"));
 		assertSame(this.mongoConverter, TestUtils.getPropertyValue(gateway, "mongoConverter"));
 		assertNotNull(TestUtils.getPropertyValue(gateway, "evaluationContext"));
-		assertTrue(TestUtils.getPropertyValue(gateway, "collectionNameExpression") instanceof SpelExpression);
+		assertThat(TestUtils.getPropertyValue(gateway, "collectionNameExpression"),
+				instanceOf(SpelExpression.class));
 		assertEquals("headers.collectionName",
 				TestUtils.getPropertyValue(gateway, "collectionNameExpression.expression"));
 	}
@@ -92,7 +96,8 @@ public class MongoDbOutboundGatewayParserTests {
 		assertSame(this.mongoDbFactory, TestUtils.getPropertyValue(gateway, "mongoDbFactory"));
 		assertSame(this.mongoConverter, TestUtils.getPropertyValue(gateway, "mongoConverter"));
 		assertNotNull(TestUtils.getPropertyValue(gateway, "evaluationContext"));
-		assertTrue(TestUtils.getPropertyValue(gateway, "collectionNameExpression") instanceof LiteralExpression);
+		assertThat(TestUtils.getPropertyValue(gateway, "collectionNameExpression"),
+				instanceOf(LiteralExpression.class));
 		assertEquals("foo", TestUtils.getPropertyValue(gateway, "collectionNameExpression.literalValue"));
 	}
 
@@ -101,14 +106,31 @@ public class MongoDbOutboundGatewayParserTests {
 		MongoDbOutboundGateway gateway = TestUtils.getPropertyValue(
 				context.getBean("fullConfigWithTemplate"), "handler", MongoDbOutboundGateway.class);
 
-		assertEquals(context.getBean("mongoDbTemplate"), TestUtils.getPropertyValue(gateway, "mongoTemplate"));
+		assertSame(context.getBean("mongoDbTemplate"), TestUtils.getPropertyValue(gateway, "mongoTemplate"));
 		assertNull(TestUtils.getPropertyValue(gateway, "mongoDbFactory"));
 		assertNull(TestUtils.getPropertyValue(gateway, "mongoConverter"));
 		assertNotNull(TestUtils.getPropertyValue(gateway, "evaluationContext"));
-		assertTrue(TestUtils.getPropertyValue(gateway, "collectionNameExpression") instanceof LiteralExpression);
+		assertThat(TestUtils.getPropertyValue(gateway, "collectionNameExpression"),
+				instanceOf(LiteralExpression.class));
 		assertEquals("foo", TestUtils.getPropertyValue(gateway, "collectionNameExpression.literalValue"));
 	}
 
+	@Test
+	public void fullConfigWithMongoDbCollectionCallback() {
+		MongoDbOutboundGateway gateway = TestUtils.getPropertyValue(
+				context.getBean("fullConfigWithMongoDbCollectionCallback"), "handler", MongoDbOutboundGateway.class);
+
+		assertSame(context.getBean("mongoDbTemplate"), TestUtils.getPropertyValue(gateway, "mongoTemplate"));
+		assertNull(TestUtils.getPropertyValue(gateway, "mongoDbFactory"));
+		assertNull(TestUtils.getPropertyValue(gateway, "mongoConverter"));
+		assertNotNull(TestUtils.getPropertyValue(gateway, "evaluationContext"));
+		assertThat(TestUtils.getPropertyValue(gateway, "collectionNameExpression"),
+				instanceOf(LiteralExpression.class));
+		assertEquals("foo", TestUtils.getPropertyValue(gateway, "collectionNameExpression.literalValue"));
+		assertThat(TestUtils.getPropertyValue(gateway, "collectionCallback"),
+				instanceOf(CollectionCallback.class));
+	}
+
 	@Test(expected = BeanDefinitionParsingException.class)
 	public void templateAndFactoryFail() {
 		new ClassPathXmlApplicationContext("outbound-gateway-fail-template-factory-config.xml", this.getClass())
@@ -121,4 +143,10 @@ public class MongoDbOutboundGatewayParserTests {
 				this.getClass()).close();
 	}
 
+	@Test(expected = BeanDefinitionParsingException.class)
+	public void collectionCallbackAndQueryFail() {
+		new ClassPathXmlApplicationContext("outbound-gateway-fail-collection-callback-config.xml",
+				this.getClass()).close();
+	}
+
 }

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/outbound-gateway-fail-collection-callback-config.xml

@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-mongodb="http://www.springframework.org/schema/integration/mongodb"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration/mongodb http://www.springframework.org/schema/integration/mongodb/spring-integration-mongodb.xsd">
+
+
+	<int:channel id="in"/>
+	<int:channel id="out"/>
+
+	<int-mongodb:outbound-gateway id="gatewayWithCollectionCallbackAndQuery"
+		mongo-template="mongoDbTemplate"
+		collection-name="foo"
+		collection-callback="mockCollectionCallback"
+		query="{'name' : 'foo'}"
+		request-channel="in"
+		reply-channel="out"/>
+
+	<bean id="mongoDbFactory" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.data.mongodb.MongoDbFactory"/>
+	</bean>
+
+	<bean id="mongoDbTemplate" class="org.springframework.data.mongodb.core.MongoTemplate">
+		<constructor-arg ref="mongoDbFactory"/>
+	</bean>
+
+	<bean id="mockCollectionCallback" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.data.mongodb.core.CollectionCallback"/>
+	</bean>
+
+</beans>

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

@@ -32,10 +32,11 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 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.MongoOperations;
-import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.BulkOperations;
 import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.CollectionCallback;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
 import org.springframework.data.mongodb.core.query.Criteria;
@@ -54,6 +55,7 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import com.mongodb.MongoClient;
+import com.mongodb.client.MongoCollection;
 
 /**
  * @author Xavier Padró
@@ -96,6 +98,10 @@ public class MongoDbTests extends MongoDbAvailableTests {
 	@Qualifier("gatewayCollectionNameFunctionFlow.input")
 	private MessageChannel gatewayCollectionNameFunctionFlow;
 
+	@Autowired
+	@Qualifier("gatewayCollectionCallbackFlow.input")
+	private MessageChannel gatewayCollectionCallbackFlow;
+
 	@Autowired
 	private MongoOperations mongoTemplate;
 
@@ -222,6 +228,20 @@ public class MongoDbTests extends MongoDbAvailableTests {
 		assertEquals("Gary", person.getName());
 	}
 
+	@Test
+	@MongoDbAvailable
+	public void testGatewayWithCollectionCallback() {
+		gatewayCollectionCallbackFlow.send(MessageBuilder
+				.withPayload("")
+				.build());
+
+		Message<?> result = this.getResultChannel.receive(10_000);
+
+		assertNotNull(result);
+		long count = (Long) result.getPayload();
+		assertEquals(4, count);
+	}
+
 	@SuppressWarnings("unchecked")
 	private List<Person> getPersons(Message<?> message) {
 		return (List<Person>) message.getPayload();
@@ -290,6 +310,13 @@ public class MongoDbTests extends MongoDbAvailableTests {
 					.channel(getResultChannel());
 		}
 
+		@Bean
+		public IntegrationFlow gatewayCollectionCallbackFlow() {
+			return f -> f
+					.handle(collectionCallbackOutboundGateway(MongoCollection::count))
+					.channel(getResultChannel());
+		}
+
 		@Bean
 		public MessageChannel getResultChannel() {
 			return MessageChannels.queue().get();
@@ -360,6 +387,13 @@ public class MongoDbTests extends MongoDbAvailableTests {
 					.entityClass(Person.class);
 		}
 
+		private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway(CollectionCallback<?> collectionCallback) {
+			return MongoDb.outboundGateway(mongoDbFactory(), mongoConverter())
+					.collectionCallback(collectionCallback)
+					.collectionName(COLLECTION_NAME)
+					.entityClass(Person.class);
+		}
+
 	}
 
 }

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbOutboundGatewayTests.java

@@ -17,8 +17,9 @@
 package org.springframework.integration.mongodb.outbound;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertNotNull;
 
 import java.util.Arrays;
 import java.util.List;
@@ -51,6 +52,8 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+import com.mongodb.client.MongoCollection;
+
 /**
  * @author Xavier Padró
  * @since 5.0
@@ -134,7 +137,7 @@ public class MongoDbOutboundGatewayTests extends MongoDbAvailableTests {
 			Assert.fail("Expected the test case to throw an IllegalArgumentException");
 		}
 		catch (IllegalStateException e) {
-			assertEquals("no query specified", e.getMessage());
+			assertEquals("no query or collectionCallback is specified", e.getMessage());
 		}
 	}
 
@@ -184,7 +187,7 @@ public class MongoDbOutboundGatewayTests extends MongoDbAvailableTests {
 			Assert.fail("Expected the test case to throw an IllegalStateException");
 		}
 		catch (IllegalStateException e) {
-			assertEquals("no query specified", e.getMessage());
+			assertEquals("no query or collectionCallback is specified", e.getMessage());
 		}
 	}
 
@@ -302,6 +305,44 @@ public class MongoDbOutboundGatewayTests extends MongoDbAvailableTests {
 		assertEquals("anotherCollection", collectionNameExpression.getValue());
 	}
 
+	@Test
+	@MongoDbAvailable
+	public void testWithCollectionCallbackCount() throws Exception {
+		Message<String> message = MessageBuilder.withPayload("").build();
+		MongoDbOutboundGateway gateway = createGateway();
+		gateway.setEntityClass(Person.class);
+		gateway.setCollectionNameExpression(new LiteralExpression("data"));
+
+		gateway.setCollectionCallback(MongoCollection::count);
+		gateway.afterPropertiesSet();
+
+		long result = (long) gateway.handleRequestMessage(message);
+
+		assertEquals(4, result);
+	}
+
+	@Test
+	@MongoDbAvailable
+	public void testWithCollectionCallbackFindOne() throws Exception {
+		Message<String> message = MessageBuilder.withPayload("").build();
+		MongoDbOutboundGateway gateway = createGateway();
+		gateway.setEntityClass(Person.class);
+		gateway.setCollectionNameExpression(new LiteralExpression("data"));
+		gateway.setRequiresReply(false);
+
+		gateway.setCollectionCallback(collection -> {
+			collection.insertOne(new Document("name", "Mike"));
+			return null;
+		});
+		gateway.afterPropertiesSet();
+
+		gateway.handleRequestMessage(message);
+
+		List<Person> persons = this.mongoTemplate.find(new Query(), Person.class, COLLECTION_NAME);
+		assertEquals(5, persons.size());
+		assertTrue(persons.stream().anyMatch(p -> p.getName().equals("Mike")));
+	}
+
 	@SuppressWarnings("unchecked")
 	private List<Person> getPersonsFromResult(Object result) {
 		return (List<Person>) result;

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbOutboundGatewayXmlTests-context.xml

@@ -64,6 +64,15 @@
 		reply-channel="out"
 		entity-class="org.springframework.integration.mongodb.rules.MongoDbAvailableTests$Person"/>
 
+	<int-mongodb:outbound-gateway id="gatewayCollectionCallback"
+		mongodb-factory="mongoDbFactory"
+		mongo-converter="mongoConverter"
+		collection-callback="countCollectionCallback"
+		collection-name-expression="headers.collectionName"
+		request-channel="in"
+		reply-channel="out"
+		entity-class="org.springframework.integration.mongodb.rules.MongoDbAvailableTests$Person"/>
+
 
 	<mongo:db-factory id="mongoDbFactory" dbname="test" />
 
@@ -79,4 +88,7 @@
 		</constructor-arg>
 	</bean>
 
+	<bean id="countCollectionCallback"
+		  class="org.springframework.integration.mongodb.rules.MongoDbAvailableTests$TestCollectionCallback" />
+
 </beans>

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbOutboundGatewayXmlTests.java

@@ -155,6 +155,24 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests {
 		assertEquals(2, persons.size());
 	}
 
+	@Test
+	@MongoDbAvailable
+	public void testCollectionCallback() throws Exception {
+		EventDrivenConsumer consumer = context.getBean("gatewayCollectionCallback", EventDrivenConsumer.class);
+		PollableChannel outChannel = context.getBean("out", PollableChannel.class);
+
+		Message<String> message = MessageBuilder
+				.withPayload("")
+				.setHeader("collectionName", "data")
+				.build();
+
+		consumer.getHandler().handleMessage(message);
+
+		Message<?> result = outChannel.receive(10000);
+		long personsCount = (Long) result.getPayload();
+		assertEquals(4, personsCount);
+	}
+
 	private Person getPerson(Message<?> message) {
 		return (Person) message.getPayload();
 	}

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -16,10 +16,14 @@
 
 package org.springframework.integration.mongodb.rules;
 
+import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.junit.Rule;
+
+import org.springframework.dao.DataAccessException;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mongodb.MongoDbFactory;
+import org.springframework.data.mongodb.core.CollectionCallback;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
 import org.springframework.data.mongodb.core.convert.DefaultDbRefResolver;
@@ -28,11 +32,14 @@ import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
 
 import com.mongodb.MongoClient;
+import com.mongodb.MongoException;
+import com.mongodb.client.MongoCollection;
 
 /**
  * Convenience base class that enables unit test methods to rely upon the {@link MongoDbAvailable} annotation.
  *
  * @author Oleg Zhurakousky
+ * @author Xavier Padró
  * @since 2.1
  */
 public abstract class MongoDbAvailableTests {
@@ -159,4 +166,13 @@ public abstract class MongoDbAvailableTests {
 
 	}
 
+	public static class TestCollectionCallback implements CollectionCallback<Long> {
+
+		@Override
+		public Long doInCollection(MongoCollection<Document> collection) throws MongoException, DataAccessException {
+			return collection.count();
+		}
+
+	}
+
 }

src/reference/asciidoc/mongodb.adoc

@@ -318,6 +318,8 @@ The Message payload and headers can be used to specify the query, as well as col
 If this attribute is not provided the default value is `org.bson.Document`;
 * `query` or `query-expression` - specifies the MongoDb query.
 Please refer to http://www.mongodb.org/display/DOCS/Querying[MongoDB documentation] for more query samples.
+* `collection-callback` - reference to an instance of `org.springframework.data.mongodb.core.CollectionCallback`
+(NOTE: you can not have both collection-callback and any of the query attributes).
 
 ==== Configuring with Java Configuration
 
@@ -337,16 +339,6 @@ public class MongoDbJavaApplication {
     @Autowired
     private MongoDbFactory mongoDbFactory;
 
-    @Bean
-    public MessageChannel requestChannel() {
-        return new DirectChannel();
-    }
-
-    @Bean
-    public MessageChannel replyChannel() {
-        return new QueueChannel(5);
-    }
-
     @Bean
     @ServiceActivator(inputChannel = "requestChannel")
     public MessageHandler mongoDbOutboundGateway() {
@@ -356,7 +348,6 @@ public class MongoDbJavaApplication {
         gateway.setExpectSingleResult(true);
         gateway.setEntityClass(Person.class);
         gateway.setOutputChannelName("replyChannel");
-
         return gateway;
     }
 
@@ -406,4 +397,17 @@ public class MongoDbJavaApplication {
     }
 
 }
+----
+
+Alternatively to the `query` and `query-expression` properties, you can specify other database operations through
+the `collectionCallback` property.
+The following example specifies a count operation:
+
+[source, java]
+----
+private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway() {
+    return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
+            .collectionCallback(MongoCollection::count)
+            .collectionName("foo");
+    }
 ----