INT-4158: Port Java DSL for JPA Module

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

* Add JavaDocs
* De-Boot `JpaDslTests`
* Fix typos and generics inconsistency in the `JpaOutboundGatewayFactoryBean`

Address PR comments

* Get rid of `JpaOutboundGatewayFactoryBean` usage in the `JpaBaseOutboundEndpointSpec`
* Rework `JpaBaseOutboundEndpointSpec` and its inheritors logic to use  `JpaOutboundGateway` directly
* Rename to the `JpaTests`
* Fix JavaDoc in the `IntegrationFlowDefinition`
* Do not use `jpaParameters` in the `JpaExecutor` if it is empty collection, not only null

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java

@@ -2761,6 +2761,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 
 	/**
 	 * Represent an Integration Flow as a Reactive Streams {@link Publisher} bean.
+	 * @param <T> the expected {@code payload} type
 	 * @return the Reactive Streams {@link Publisher}
 	 */
 	@SuppressWarnings("unchecked")

spring-integration-jpa/src/main/java/org/springframework/integration/jpa/core/JpaExecutor.java

@@ -39,6 +39,7 @@ import org.springframework.integration.jpa.support.parametersource.ParameterSour
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 
 /**
  * Executes Jpa Operations that produce payload objects from the result of the provided:
@@ -172,7 +173,7 @@ public class JpaExecutor implements InitializingBean, BeanFactoryAware {
 	@Override
 	public void afterPropertiesSet() {
 
-		if (this.jpaParameters != null) {
+		if (!CollectionUtils.isEmpty(this.jpaParameters)) {
 
 			if (this.parameterSourceFactory == null) {
 				ExpressionEvaluatingParameterSourceFactory expressionSourceFactory =
@@ -597,18 +598,16 @@ public class JpaExecutor implements InitializingBean, BeanFactoryAware {
 		this.firstResultExpression = firstResultExpression;
 	}
 
-
 	/**
 	 * Set the expression that will be evaluated to get the {@code primaryKey} for
 	 * {@link javax.persistence.EntityManager#find(Class, Object)}
-	 * @param idExpression The first result expression.
+	 * @param idExpression the SpEL expression for entity {@code primaryKey}.
 	 * @since 4.0
 	 */
 	public void setIdExpression(Expression idExpression) {
 		this.idExpression = idExpression;
 	}
 
-
 	/**
 	 * Set the expression for maximum number of results expression. It has be a non null value
 	 * Not setting one will default to the behavior of fetching all the records

spring-integration-jpa/src/main/java/org/springframework/integration/jpa/dsl/Jpa.java

@@ -0,0 +1,176 @@
+/*
+ * Copyright 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.
+ * 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.jpa.dsl;
+
+import javax.persistence.EntityManager;
+import javax.persistence.EntityManagerFactory;
+
+import org.springframework.integration.jpa.core.JpaExecutor;
+import org.springframework.integration.jpa.core.JpaOperations;
+
+/**
+ * Factory class for JPA components.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public final class Jpa {
+
+	/**
+	 * Create a {@link JpaInboundChannelAdapterSpec} builder instance
+	 * based on the provided {@link EntityManagerFactory}.
+	 * @param entityManagerFactory the {@link EntityManagerFactory} to use
+	 * @return the JpaInboundChannelAdapterSpec instance
+	 */
+	public static JpaInboundChannelAdapterSpec inboundAdapter(EntityManagerFactory entityManagerFactory) {
+		return inboundAdapter(new JpaExecutor(entityManagerFactory));
+	}
+
+	/**
+	 * Create a {@link JpaInboundChannelAdapterSpec} builder instance
+	 * based on the provided {@link EntityManager}.
+	 * @param entityManager the {@link EntityManager} to use
+	 * @return the JpaInboundChannelAdapterSpec instance
+	 */
+	public static JpaInboundChannelAdapterSpec inboundAdapter(EntityManager entityManager) {
+		return inboundAdapter(new JpaExecutor(entityManager));
+	}
+
+	/**
+	 * Create a {@link JpaInboundChannelAdapterSpec} builder instance
+	 * based on the provided {@link JpaOperations}.
+	 * @param jpaOperations the {@link JpaOperations} to use
+	 * @return the JpaInboundChannelAdapterSpec instance
+	 */
+	public static JpaInboundChannelAdapterSpec inboundAdapter(JpaOperations jpaOperations) {
+		return inboundAdapter(new JpaExecutor(jpaOperations));
+	}
+
+	private static JpaInboundChannelAdapterSpec inboundAdapter(JpaExecutor jpaExecutor) {
+		return new JpaInboundChannelAdapterSpec(jpaExecutor);
+	}
+
+	/**
+	 * Create a {@link JpaUpdatingOutboundEndpointSpec} builder instance for one-way adapter
+	 * based on the provided {@link EntityManagerFactory}.
+	 * @param entityManagerFactory the {@link EntityManagerFactory} to use
+	 * @return the JpaUpdatingOutboundEndpointSpec instance
+	 */
+	public static JpaUpdatingOutboundEndpointSpec outboundAdapter(EntityManagerFactory entityManagerFactory) {
+		return outboundAdapter(new JpaExecutor(entityManagerFactory));
+	}
+
+	/**
+	 * Create a {@link JpaUpdatingOutboundEndpointSpec} builder instance for one-way adapter
+	 * based on the provided {@link EntityManager}.
+	 * @param entityManager the {@link EntityManager} to use
+	 * @return the JpaUpdatingOutboundEndpointSpec instance
+	 */
+	public static JpaUpdatingOutboundEndpointSpec outboundAdapter(EntityManager entityManager) {
+		return outboundAdapter(new JpaExecutor(entityManager));
+	}
+
+	/**
+	 * Create a {@link JpaUpdatingOutboundEndpointSpec} builder instance for one-way adapter
+	 * based on the provided {@link JpaOperations}.
+	 * @param jpaOperations the {@link JpaOperations} to use
+	 * @return the JpaUpdatingOutboundEndpointSpec instance
+	 */
+	public static JpaUpdatingOutboundEndpointSpec outboundAdapter(JpaOperations jpaOperations) {
+		return outboundAdapter(new JpaExecutor(jpaOperations));
+	}
+
+	private static JpaUpdatingOutboundEndpointSpec outboundAdapter(JpaExecutor jpaExecutor) {
+		return new JpaUpdatingOutboundEndpointSpec(jpaExecutor)
+				.producesReply(false);
+	}
+
+	/**
+	 * Create a {@link JpaUpdatingOutboundEndpointSpec} builder instance for request-reply gateway
+	 * based on the provided {@link EntityManagerFactory}.
+	 * @param entityManagerFactory the {@link EntityManagerFactory} to use
+	 * @return the JpaUpdatingOutboundEndpointSpec instance
+	 */
+	public static JpaUpdatingOutboundEndpointSpec updatingGateway(EntityManagerFactory entityManagerFactory) {
+		return updatingGateway(new JpaExecutor(entityManagerFactory));
+	}
+
+	/**
+	 * Create a {@link JpaUpdatingOutboundEndpointSpec} builder instance for request-reply gateway
+	 * based on the provided {@link EntityManager}.
+	 * @param entityManager the {@link EntityManager} to use
+	 * @return the JpaUpdatingOutboundEndpointSpec instance
+	 */
+	public static JpaUpdatingOutboundEndpointSpec updatingGateway(EntityManager entityManager) {
+		return updatingGateway(new JpaExecutor(entityManager));
+	}
+
+	/**
+	 * Create a {@link JpaUpdatingOutboundEndpointSpec} builder instance for request-reply gateway
+	 * based on the provided {@link JpaOperations}.
+	 * @param jpaOperations the {@link JpaOperations} to use
+	 * @return the JpaUpdatingOutboundEndpointSpec instance
+	 */
+	public static JpaUpdatingOutboundEndpointSpec updatingGateway(JpaOperations jpaOperations) {
+		return updatingGateway(new JpaExecutor(jpaOperations));
+	}
+
+	private static JpaUpdatingOutboundEndpointSpec updatingGateway(JpaExecutor jpaExecutor) {
+		return new JpaUpdatingOutboundEndpointSpec(jpaExecutor)
+				.producesReply(true);
+	}
+
+	/**
+	 * Create a {@link JpaRetrievingOutboundGatewaySpec} builder instance for request-reply gateway
+	 * based on the provided {@link EntityManagerFactory}.
+	 * @param entityManagerFactory the {@link EntityManagerFactory} to use
+	 * @return the JpaRetrievingOutboundGatewaySpec instance
+	 */
+	public static JpaRetrievingOutboundGatewaySpec retrievingGateway(EntityManagerFactory entityManagerFactory) {
+		return retrievingGateway(new JpaExecutor(entityManagerFactory));
+	}
+
+	/**
+	 * Create a {@link JpaRetrievingOutboundGatewaySpec} builder instance for request-reply gateway
+	 * based on the provided {@link EntityManager}.
+	 * @param entityManager the {@link EntityManager} to use
+	 * @return the JpaRetrievingOutboundGatewaySpec instance
+	 */
+	public static JpaRetrievingOutboundGatewaySpec retrievingGateway(EntityManager entityManager) {
+		return retrievingGateway(new JpaExecutor(entityManager));
+	}
+
+	/**
+	 * Create a {@link JpaRetrievingOutboundGatewaySpec} builder instance for request-reply gateway
+	 * based on the provided {@link JpaOperations}.
+	 * @param jpaOperations the {@link JpaOperations} to use
+	 * @return the JpaRetrievingOutboundGatewaySpec instance
+	 */
+	public static JpaRetrievingOutboundGatewaySpec retrievingGateway(JpaOperations jpaOperations) {
+		return retrievingGateway(new JpaExecutor(jpaOperations));
+	}
+
+	private static JpaRetrievingOutboundGatewaySpec retrievingGateway(JpaExecutor jpaExecutor) {
+		return new JpaRetrievingOutboundGatewaySpec(jpaExecutor);
+	}
+
+	private Jpa() {
+		super();
+	}
+
+}

spring-integration-jpa/src/main/java/org/springframework/integration/jpa/dsl/JpaBaseOutboundEndpointSpec.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 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.
+ * 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.jpa.dsl;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.springframework.integration.dsl.ComponentsRegistration;
+import org.springframework.integration.dsl.MessageHandlerSpec;
+import org.springframework.integration.jpa.core.JpaExecutor;
+import org.springframework.integration.jpa.outbound.JpaOutboundGateway;
+import org.springframework.integration.jpa.support.JpaParameter;
+import org.springframework.integration.jpa.support.parametersource.BeanPropertyParameterSourceFactory;
+import org.springframework.integration.jpa.support.parametersource.ParameterSourceFactory;
+
+/**
+ * The base {@link MessageHandlerSpec} for JPA Outbound endpoints.
+ *
+ * @param <S> the target {@link JpaBaseOutboundEndpointSpec} implementation type.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public abstract class JpaBaseOutboundEndpointSpec<S extends JpaBaseOutboundEndpointSpec<S>>
+		extends MessageHandlerSpec<S, JpaOutboundGateway>
+		implements ComponentsRegistration {
+
+	private final List<JpaParameter> jpaParameters = new LinkedList<>();
+
+	protected final JpaExecutor jpaExecutor;
+
+	protected JpaBaseOutboundEndpointSpec(JpaExecutor jpaExecutor) {
+		this.jpaExecutor = jpaExecutor;
+		this.jpaExecutor.setJpaParameters(this.jpaParameters);
+		this.target = new JpaOutboundGateway(this.jpaExecutor);
+	}
+
+	/**
+	 * Specify the class type which is being used for retrieving entities from the database.
+	 * @param entityClass the entity {@link Class} to use
+	 * @return the spec
+	 */
+	public S entityClass(Class<?> entityClass) {
+		this.jpaExecutor.setEntityClass(entityClass);
+		return _this();
+	}
+
+	/**
+	 * Specify a JPA query to perform persistent operation.
+	 * @param jpaQuery the JPA query to use.
+	 * @return the spec
+	 */
+	public S jpaQuery(String jpaQuery) {
+		this.jpaExecutor.setJpaQuery(jpaQuery);
+		return _this();
+	}
+
+	/**
+	 * Specify a native SQL query to perform persistent operation.
+	 * @param nativeQuery the native SQL query to use.
+	 * @return the spec
+	 */
+	public S nativeQuery(String nativeQuery) {
+		this.jpaExecutor.setNativeQuery(nativeQuery);
+		return _this();
+	}
+
+	/**
+	 * Specify a name a named JPQL based query or a native SQL query.
+	 * @param namedQuery the name of the pre-configured query.
+	 * @return the spec
+	 */
+	public S namedQuery(String namedQuery) {
+		this.jpaExecutor.setNamedQuery(namedQuery);
+		return _this();
+	}
+
+	/**
+	 * Specify a {@link ParameterSourceFactory} to populate query parameters at runtime against request message.
+	 * @param parameterSourceFactory the {@link ParameterSourceFactory} to use.
+	 * @return the spec
+	 */
+	public S parameterSourceFactory(ParameterSourceFactory parameterSourceFactory) {
+		this.jpaExecutor.setParameterSourceFactory(parameterSourceFactory);
+		return _this();
+	}
+
+	/**
+	 * Add a value for indexed query parameter.
+	 * @param value the value for query parameter by index
+	 * @return the spec
+	 */
+	public S parameter(Object value) {
+		return parameter(new JpaParameter(value, null));
+	}
+
+	/**
+	 * Add a value for named parameter in the query.
+	 * @param name the name of the query parameter
+	 * @param value the value for query parameter by name
+	 * @return the spec
+	 */
+	public S parameter(String name, Object value) {
+		return parameter(new JpaParameter(name, value, null));
+	}
+
+	/**
+	 * Add a SpEL expression for indexed parameter in the query.
+	 * @param expression the SpEL expression for query parameter by index
+	 * @return the spec
+	 */
+	public S parameterExpression(String expression) {
+		return parameter(new JpaParameter(null, expression));
+	}
+
+	/**
+	 * Add a SpEL expression for named parameter in the query.
+	 * @param name the name of the query parameter
+	 * @param expression the SpEL expression for query parameter by name
+	 * @return the spec
+	 */
+	public S parameterExpression(String name, String expression) {
+		return parameter(new JpaParameter(name, null, expression));
+	}
+
+	public S parameter(JpaParameter jpaParameter) {
+		this.jpaParameters.add(jpaParameter);
+		return _this();
+	}
+
+	/**
+	 * Indicates that whether only the payload of the passed in {@code Message}
+	 * will be used as a source of parameters. The is 'true' by default because as a
+	 * default a {@link BeanPropertyParameterSourceFactory} implementation is
+	 * used for the sqlParameterSourceFactory property.
+	 * @param usePayloadAsParameterSource the {@code boolean} flag to indicate
+	 * if use {@code payload} as a source of parameter values or not.
+	 * @return the spec
+	 */
+	public S usePayloadAsParameterSource(Boolean usePayloadAsParameterSource) {
+		this.jpaExecutor.setUsePayloadAsParameterSource(usePayloadAsParameterSource);
+		return _this();
+	}
+
+	@Override
+	public Collection<Object> getComponentsToRegister() {
+		return Collections.singletonList(this.jpaExecutor);
+	}
+
+}

spring-integration-jpa/src/main/java/org/springframework/integration/jpa/dsl/JpaInboundChannelAdapterSpec.java

@@ -0,0 +1,190 @@
+/*
+ * Copyright 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.
+ * 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.jpa.dsl;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import org.springframework.expression.Expression;
+import org.springframework.integration.dsl.ComponentsRegistration;
+import org.springframework.integration.dsl.MessageSourceSpec;
+import org.springframework.integration.expression.ValueExpression;
+import org.springframework.integration.jpa.core.JpaExecutor;
+import org.springframework.integration.jpa.inbound.JpaPollingChannelAdapter;
+import org.springframework.integration.jpa.support.parametersource.ParameterSource;
+
+/**
+ * A {@link MessageSourceSpec} for a {@link JpaPollingChannelAdapter}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public class JpaInboundChannelAdapterSpec
+		extends MessageSourceSpec<JpaInboundChannelAdapterSpec, JpaPollingChannelAdapter>
+		implements ComponentsRegistration {
+
+	private final JpaExecutor jpaExecutor;
+
+	JpaInboundChannelAdapterSpec(JpaExecutor jpaExecutor) {
+		this.jpaExecutor = jpaExecutor;
+		this.target = new JpaPollingChannelAdapter(this.jpaExecutor);
+	}
+
+	/**
+	 * Specify the class type which is being used for retrieving entities from the database.
+	 * @param entityClass the entity {@link Class} to use
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec entityClass(Class<?> entityClass) {
+		this.jpaExecutor.setEntityClass(entityClass);
+		return this;
+	}
+
+	/**
+	 * Specify a JPA query to perform persistent operation.
+	 * @param jpaQuery the JPA query to use.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec jpaQuery(String jpaQuery) {
+		this.jpaExecutor.setJpaQuery(jpaQuery);
+		return this;
+	}
+
+	/**
+	 * Specify a native SQL query to perform persistent operation.
+	 * @param nativeQuery the native SQL query to use.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec nativeQuery(String nativeQuery) {
+		this.jpaExecutor.setNativeQuery(nativeQuery);
+		return this;
+	}
+
+	/**
+	 * Specify a name a named JPQL based query or a native SQL query.
+	 * @param namedQuery the name of the pre-configured query.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec namedQuery(String namedQuery) {
+		this.jpaExecutor.setNamedQuery(namedQuery);
+		return this;
+	}
+
+	/**
+	 * If set to 'true', the retrieved objects are deleted from the database upon
+	 * being polled. May not work in all situations, e.g. for Native SQL Queries.
+	 * @param deleteAfterPoll Defaults to 'false'.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec deleteAfterPoll(boolean deleteAfterPoll) {
+		this.jpaExecutor.setDeleteAfterPoll(deleteAfterPoll);
+		return this;
+	}
+
+	/**
+	 * If not set, this property defaults to <code>false</code>, which means that
+	 * deletion occurs on a per object basis if a collection of entities is being
+	 * deleted.
+	 *<p>If set to 'true' the elements of the payload are deleted as a batch
+	 * operation. Be aware that this exhibits issues in regards to cascaded deletes.
+	 *<p>The specification 'JSR 317: Java Persistence API, Version 2.0' does not
+	 * support cascaded deletes in batch operations. The specification states in
+	 * chapter 4.10:
+	 *<p>"A delete operation only applies to entities of the specified class and
+	 * its subclasses. It does not cascade to related entities."
+	 * @param deleteInBatch Defaults to 'false' if not set.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec deleteInBatch(boolean deleteInBatch) {
+		this.jpaExecutor.setDeleteInBatch(deleteInBatch);
+		return this;
+	}
+
+	/**
+	 * If set to {@code true} the {@link javax.persistence.EntityManager#flush()} will be called
+	 * after persistence operation.
+	 * Has the same effect, if the {@code flushSize} is specified to {@code 1}.
+	 * For convenience in cases when the provided entity to persist is not an instance of {@link Iterable}.
+	 * @param flush defaults to 'false'.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec flushAfterDelete(boolean flush) {
+		this.jpaExecutor.setFlush(flush);
+		return this;
+	}
+
+	/**
+	 * Specify a {@link ParameterSource} that would be used to provide additional parameters.
+	 * @param parameterSource the {@link ParameterSource} to use.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec parameterSource(ParameterSource parameterSource) {
+		this.jpaExecutor.setParameterSource(parameterSource);
+		return this;
+	}
+
+	/**
+	 * This parameter indicates that only one result object shall be returned as
+	 * a result from the executed JPA operation. If set to <code>true</code> and
+	 * the result list from the JPA operations contains only 1 element, then that
+	 * 1 element is extracted and returned as payload.
+	 * @param expectSingleResult true if a single object is expected.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec expectSingleResult(boolean expectSingleResult) {
+		this.jpaExecutor.setExpectSingleResult(expectSingleResult);
+		return this;
+	}
+
+	/**
+	 * Set the maximum number of results expression. It has be a non null value
+	 * Not setting one will default to the behavior of fetching all the records
+	 * @param maxResults the maximum number of results to retrieve
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec maxResults(int maxResults) {
+		return maxResultsExpression(new ValueExpression<>(maxResults));
+	}
+
+	/**
+	 * Specify a SpEL expression for maximum number of results expression.
+	 * Not setting one will default to the behavior of fetching all the records
+	 * @param maxResultsExpression The maximum results expression.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec maxResultsExpression(String maxResultsExpression) {
+		return maxResultsExpression(PARSER.parseExpression(maxResultsExpression));
+	}
+
+	/**
+	 * Specify a SpEL expression for maximum number of results expression.
+	 * Not setting one will default to the behavior of fetching all the records
+	 * @param maxResultsExpression The maximum results expression.
+	 * @return the spec
+	 */
+	public JpaInboundChannelAdapterSpec maxResultsExpression(Expression maxResultsExpression) {
+		this.jpaExecutor.setMaxResultsExpression(maxResultsExpression);
+		return this;
+	}
+
+	@Override
+	public Collection<Object> getComponentsToRegister() {
+		return Collections.singletonList(this.jpaExecutor);
+	}
+
+}

spring-integration-jpa/src/main/java/org/springframework/integration/jpa/dsl/JpaRetrievingOutboundGatewaySpec.java

@@ -0,0 +1,177 @@
+/*
+ * Copyright 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.
+ * 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.jpa.dsl;
+
+import org.springframework.expression.Expression;
+import org.springframework.integration.expression.ValueExpression;
+import org.springframework.integration.jpa.core.JpaExecutor;
+import org.springframework.integration.jpa.support.OutboundGatewayType;
+
+/**
+ * A {@link JpaBaseOutboundEndpointSpec} extension for the
+ * {@link org.springframework.integration.jpa.outbound.JpaOutboundGateway} with
+ * {@link org.springframework.integration.jpa.support.OutboundGatewayType#RETRIEVING} mode.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public class JpaRetrievingOutboundGatewaySpec extends JpaBaseOutboundEndpointSpec<JpaRetrievingOutboundGatewaySpec> {
+
+	JpaRetrievingOutboundGatewaySpec(JpaExecutor jpaExecutor) {
+		super(jpaExecutor);
+		this.target.setGatewayType(OutboundGatewayType.RETRIEVING);
+		this.target.setRequiresReply(true);
+	}
+
+	/**
+	 * This parameter indicates that only one result object shall be returned as
+	 * a result from the executed JPA operation. If set to <code>true</code> and
+	 * the result list from the JPA operations contains only 1 element, then that
+	 * 1 element is extracted and returned as payload.
+	 * @param expectSingleResult true if a single object is expected.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec expectSingleResult(boolean expectSingleResult) {
+		this.jpaExecutor.setExpectSingleResult(expectSingleResult);
+		return this;
+	}
+
+	/**
+	 * Specify a first result in the query executed.
+	 * @param firstResult the first result to use.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec firstResult(int firstResult) {
+		return firstResultExpression(new ValueExpression<>(firstResult));
+	}
+
+	/**
+	 * Specify a SpEL expression that will be evaluated to get the first result in the query executed.
+	 * @param firstResultExpression The first result expression.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec firstResultExpression(String firstResultExpression) {
+		return firstResultExpression(PARSER.parseExpression(firstResultExpression));
+	}
+
+	/**
+	 * Specify a SpEL expression that will be evaluated to get the first result in the query executed.
+	 * @param firstResultExpression The first result expression.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec firstResultExpression(Expression firstResultExpression) {
+		this.jpaExecutor.setFirstResultExpression(firstResultExpression);
+		return this;
+	}
+
+	/**
+	 * Specify a SpEL expression that will be evaluated to get the {@code primaryKey} for
+	 * {@link javax.persistence.EntityManager#find(Class, Object)}
+	 * @param idExpression the SpEL expression for entity {@code primaryKey}.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec idExpression(String idExpression) {
+		return idExpression(PARSER.parseExpression(idExpression));
+	}
+
+	/**
+	 * Specify a SpEL expression that will be evaluated to get the {@code primaryKey} for
+	 * {@link javax.persistence.EntityManager#find(Class, Object)}
+	 * @param idExpression the SpEL expression for entity {@code primaryKey}.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec idExpression(Expression idExpression) {
+		this.jpaExecutor.setIdExpression(idExpression);
+		return this;
+	}
+
+	/**
+	 * Set the maximum number of results expression. It has be a non null value
+	 * Not setting one will default to the behavior of fetching all the records
+	 * @param maxResults the maximum number of results to retrieve
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec maxResults(int maxResults) {
+		return maxResultsExpression(new ValueExpression<>(maxResults));
+	}
+
+	/**
+	 * Specify a SpEL expression for maximum number of results expression.
+	 * Not setting one will default to the behavior of fetching all the records
+	 * @param maxResultsExpression The maximum results expression.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec maxResultsExpression(String maxResultsExpression) {
+		return maxResultsExpression(PARSER.parseExpression(maxResultsExpression));
+	}
+
+	/**
+	 * Specify a SpEL expression for maximum number of results expression.
+	 * Not setting one will default to the behavior of fetching all the records
+	 * @param maxResultsExpression The maximum results expression.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec maxResultsExpression(Expression maxResultsExpression) {
+		this.jpaExecutor.setMaxResultsExpression(maxResultsExpression);
+		return this;
+	}
+
+	/**
+	 * If set to {@code true}, the retrieved objects are deleted from the database upon
+	 * being polled. May not work in all situations, e.g. for Native SQL Queries.
+	 * @param deleteAfterPoll defaults to {@code false}.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec deleteAfterPoll(boolean deleteAfterPoll) {
+		this.jpaExecutor.setDeleteAfterPoll(deleteAfterPoll);
+		return this;
+	}
+
+	/**
+	 * If not set, this property defaults to <code>false</code>, which means that
+	 * deletion occurs on a per object basis if a collection of entities is being
+	 * deleted.
+	 *<p>If set to 'true' the elements of the payload are deleted as a batch
+	 * operation. Be aware that this exhibits issues in regards to cascaded deletes.
+	 *<p>The specification 'JSR 317: Java Persistence API, Version 2.0' does not
+	 * support cascaded deletes in batch operations. The specification states in
+	 * chapter 4.10:
+	 *<p>"A delete operation only applies to entities of the specified class and
+	 * its subclasses. It does not cascade to related entities."
+	 * @param deleteInBatch Defaults to 'false' if not set.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec deleteInBatch(boolean deleteInBatch) {
+		this.jpaExecutor.setDeleteInBatch(deleteInBatch);
+		return this;
+	}
+
+	/**
+	 * If set to {@code true} the {@link javax.persistence.EntityManager#flush()} will be called
+	 * after persistence operation.
+	 * Has the same effect, if the {@code flushSize} is specified to {@code 1}.
+	 * For convenience in cases when the provided entity to persist is not an instance of {@link Iterable}.
+	 * @param flush defaults to 'false'.
+	 * @return the spec
+	 */
+	public JpaRetrievingOutboundGatewaySpec flushAfterDelete(boolean flush) {
+		this.jpaExecutor.setFlush(flush);
+		return this;
+	}
+
+}

spring-integration-jpa/src/main/java/org/springframework/integration/jpa/dsl/JpaUpdatingOutboundEndpointSpec.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 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.
+ * 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.jpa.dsl;
+
+import org.springframework.integration.jpa.core.JpaExecutor;
+import org.springframework.integration.jpa.support.PersistMode;
+
+/**
+ * A {@link JpaBaseOutboundEndpointSpec} extension for the {@code updating}
+ * {@link org.springframework.integration.jpa.outbound.JpaOutboundGateway} mode.
+ * The {@code outbound-channel-adapter} is achievable through an internal {@code producesReply} option.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public class JpaUpdatingOutboundEndpointSpec extends JpaBaseOutboundEndpointSpec<JpaUpdatingOutboundEndpointSpec> {
+
+	JpaUpdatingOutboundEndpointSpec(JpaExecutor jpaExecutor) {
+		super(jpaExecutor);
+	}
+
+	JpaUpdatingOutboundEndpointSpec producesReply(boolean producesReply) {
+		this.target.setProducesReply(producesReply);
+		if (producesReply) {
+			this.target.setRequiresReply(true);
+		}
+		return this;
+	}
+
+	/**
+	 * Specify a {@link PersistMode} for the gateway.
+	 * Defaults to {@link PersistMode#MERGE}.
+	 * @param persistMode the {@link PersistMode} to use.
+	 * @return the spec
+	 */
+	public JpaUpdatingOutboundEndpointSpec persistMode(PersistMode persistMode) {
+		this.jpaExecutor.setPersistMode(persistMode);
+		return this;
+	}
+
+	/**
+	 * If set to {@code true} the {@link javax.persistence.EntityManager#flush()} will be called
+	 * after persistence operation.
+	 * Has the same effect, if the {@link #flushSize} is specified to {@code 1}.
+	 * For convenience in cases when the provided entity to persist is not an instance of {@link Iterable}.
+	 * @param flush defaults to {@code false}.
+	 * @return the spec
+	 */
+	public JpaUpdatingOutboundEndpointSpec flush(boolean flush) {
+		this.jpaExecutor.setFlush(flush);
+		return this;
+	}
+
+	/**
+	 * If the provided value is greater than {@code 0}, then {@link javax.persistence.EntityManager#flush()}
+	 * will be called after persistence operations as well as within batch operations.
+	 * This property has precedence over the {@link #flush}, if it is specified to a value greater than {@code 0}.
+	 * If the entity to persist is not an instance of {@link Iterable} and this property is greater than {@code 0},
+	 * then the entity will be flushed as if the {@link #flush} attribute was set to {@code true}.
+	 * @param flushSize defaults to {@code 0}.
+	 * @return the spec
+	 */
+	public JpaUpdatingOutboundEndpointSpec flushSize(int flushSize) {
+		this.jpaExecutor.setFlushSize(flushSize);
+		return this;
+	}
+
+	/**
+	 * If set to {@code true} the {@link javax.persistence.EntityManager#clear()} will be called,
+	 * and only if the {@link javax.persistence.EntityManager#flush()}
+	 * was called after performing persistence operations.
+	 * @param clearOnFlush defaults to {@code false}.
+	 * @return the spec
+	 */
+	public JpaUpdatingOutboundEndpointSpec clearOnFlush(boolean clearOnFlush) {
+		this.jpaExecutor.setClearOnFlush(clearOnFlush);
+		return this;
+	}
+
+}

spring-integration-jpa/src/main/java/org/springframework/integration/jpa/dsl/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides JPA Components support for Java DSL.
+ */
+package org.springframework.integration.jpa.dsl;

spring-integration-jpa/src/main/java/org/springframework/integration/jpa/outbound/JpaOutboundGatewayFactoryBean.java

@@ -41,7 +41,7 @@ import org.springframework.transaction.interceptor.TransactionInterceptor;
  * @since 2.2
  *
  */
-public class JpaOutboundGatewayFactoryBean extends AbstractFactoryBean<MessageHandler> {
+public class JpaOutboundGatewayFactoryBean extends AbstractFactoryBean<JpaOutboundGateway> {
 
 	private JpaExecutor jpaExecutor;
 
@@ -120,7 +120,7 @@ public class JpaOutboundGatewayFactoryBean extends AbstractFactoryBean<MessageHa
 	}
 
 	@Override
-	protected MessageHandler createInstance() {
+	protected JpaOutboundGateway createInstance() {
 		JpaOutboundGateway jpaOutboundGateway = new JpaOutboundGateway(this.jpaExecutor);
 		jpaOutboundGateway.setGatewayType(this.gatewayType);
 		jpaOutboundGateway.setProducesReply(this.producesReply);

spring-integration-jpa/src/test/java/org/springframework/integration/jpa/dsl/JpaTests.java

@@ -0,0 +1,261 @@
+/*
+ * Copyright 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.
+ * 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.jpa.dsl;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Calendar;
+import java.util.Date;
+import java.util.List;
+
+import javax.persistence.EntityManagerFactory;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.jpa.support.PersistMode;
+import org.springframework.integration.jpa.test.entity.Gender;
+import org.springframework.integration.jpa.test.entity.StudentDomain;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.OnlyOnceTrigger;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.orm.jpa.JpaTransactionManager;
+import org.springframework.orm.jpa.JpaVendorAdapter;
+import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;
+import org.springframework.orm.jpa.vendor.AbstractJpaVendorAdapter;
+import org.springframework.orm.jpa.vendor.Database;
+import org.springframework.orm.jpa.vendor.HibernateJpaVendorAdapter;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.transaction.PlatformTransactionManager;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class JpaTests {
+
+	private static EmbeddedDatabase dataSource;
+
+	@Autowired
+	private PollableChannel pollingResults;
+
+	@Autowired
+	@Qualifier("outboundAdapterFlow.input")
+	private MessageChannel outboundAdapterFlowInput;
+
+	@Autowired
+	@Qualifier("updatingGatewayFlow.input")
+	private MessageChannel updatingGatewayFlowInput;
+
+	@Autowired
+	private PollableChannel persistResults;
+
+	@Autowired
+	@Qualifier("retrievingGatewayFlow.input")
+	private MessageChannel retrievingGatewayFlowInput;
+
+
+	@Autowired
+	private PollableChannel retrieveResults;
+
+	@BeforeClass
+	public static void init() {
+		dataSource = new EmbeddedDatabaseBuilder()
+				.setType(EmbeddedDatabaseType.H2)
+				.addScript("classpath:H2-DropTables.sql")
+				.addScript("classpath:H2-CreateTables.sql")
+				.addScript("classpath:H2-PopulateData.sql")
+				.ignoreFailedDrops(true)
+				.build();
+	}
+
+	@AfterClass
+	public static void destroy() {
+		dataSource.shutdown();
+	}
+
+	@Test
+	public void testInboundAdapterFlow() {
+		Message<?> message = this.pollingResults.receive(10_000);
+		assertNotNull(message);
+		assertThat(message.getPayload(), instanceOf(StudentDomain.class));
+		StudentDomain student = (StudentDomain) message.getPayload();
+		assertEquals("First One", student.getFirstName());
+	}
+
+	@Test
+	public void testOutboundAdapterFlow() {
+		JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
+
+		List<?> results1 = jdbcTemplate.queryForList("Select * from Student");
+		assertNotNull(results1);
+		assertTrue(results1.size() == 3);
+
+		Calendar dateOfBirth = Calendar.getInstance();
+		dateOfBirth.set(1981, 9, 27);
+
+		StudentDomain student = new StudentDomain()
+				.withFirstName("Artem")
+				.withLastName("Bilan")
+				.withGender(Gender.MALE)
+				.withDateOfBirth(dateOfBirth.getTime())
+				.withLastUpdated(new Date());
+
+		assertNull(student.getRollNumber());
+
+		this.outboundAdapterFlowInput.send(MessageBuilder.withPayload(student).build());
+
+		List<?> results2 = jdbcTemplate.queryForList("Select * from Student");
+		assertNotNull(results2);
+		assertTrue(results2.size() == 4);
+
+		assertNotNull(student.getRollNumber());
+	}
+
+	@Test
+	public void testUpdatingGatewayFlow() {
+		Calendar dateOfBirth = Calendar.getInstance();
+		dateOfBirth.set(1981, 9, 27);
+
+		StudentDomain student = new StudentDomain()
+				.withFirstName("Artem")
+				.withLastName("Bilan")
+				.withGender(Gender.MALE)
+				.withDateOfBirth(dateOfBirth.getTime())
+				.withLastUpdated(new Date());
+
+		assertNull(student.getRollNumber());
+
+		this.updatingGatewayFlowInput.send(MessageBuilder.withPayload(student).build());
+
+		Message<?> receive = this.persistResults.receive(10_000);
+		assertNotNull(receive);
+
+		StudentDomain mergedStudent = (StudentDomain) receive.getPayload();
+		assertEquals(student.getFirstName(), mergedStudent.getFirstName());
+		assertNotNull(mergedStudent.getRollNumber());
+		assertNull(student.getRollNumber());
+	}
+
+	@Test
+	public void testRetrievingGatewayFlow() {
+		this.retrievingGatewayFlowInput.send(MessageBuilder.withPayload(1002L).build());
+		Message<?> receive = this.retrieveResults.receive(10_000);
+		assertNotNull(receive);
+		assertThat(receive.getPayload(), instanceOf(StudentDomain.class));
+		StudentDomain student = (StudentDomain) receive.getPayload();
+		assertEquals("First Two", student.getFirstName());
+		assertEquals(Gender.FEMALE, student.getGender());
+	}
+
+
+	@Configuration
+	@EnableIntegration
+	public static class ContextConfiguration {
+
+		@Bean
+		public JpaVendorAdapter jpaVendorAdapter() {
+			AbstractJpaVendorAdapter adapter = new HibernateJpaVendorAdapter();
+			adapter.setShowSql(true);
+			adapter.setDatabase(Database.H2);
+			adapter.setGenerateDdl(true);
+			return adapter;
+		}
+
+		@Bean
+		public LocalContainerEntityManagerFactoryBean entityManagerFactory() {
+			LocalContainerEntityManagerFactoryBean entityManagerFactoryBean =
+					new LocalContainerEntityManagerFactoryBean();
+			entityManagerFactoryBean.setDataSource(dataSource);
+			entityManagerFactoryBean.setPersistenceUnitName("persistenceUnit");
+			entityManagerFactoryBean.setJpaVendorAdapter(jpaVendorAdapter());
+			return entityManagerFactoryBean;
+		}
+
+		@Bean
+		public PlatformTransactionManager transactionManager(EntityManagerFactory entityManagerFactory) {
+			JpaTransactionManager jpaTransactionManager = new JpaTransactionManager(entityManagerFactory);
+			jpaTransactionManager.setDataSource(dataSource);
+			return jpaTransactionManager;
+		}
+
+		@Bean
+		public IntegrationFlow pollingAdapterFlow(EntityManagerFactory entityManagerFactory) {
+			return IntegrationFlows
+					.from(Jpa.inboundAdapter(entityManagerFactory)
+									.entityClass(StudentDomain.class)
+									.maxResults(1)
+									.expectSingleResult(true),
+							e -> e.poller(p -> p.trigger(new OnlyOnceTrigger())))
+					.channel(c -> c.queue("pollingResults"))
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow outboundAdapterFlow(EntityManagerFactory entityManagerFactory) {
+			return f -> f
+					.handle(Jpa.outboundAdapter(entityManagerFactory)
+									.entityClass(StudentDomain.class)
+									.persistMode(PersistMode.PERSIST),
+							e -> e.transactional(true));
+		}
+
+		@Bean
+		public IntegrationFlow updatingGatewayFlow(EntityManagerFactory entityManagerFactory) {
+			return f -> f
+					.handle(Jpa.updatingGateway(entityManagerFactory),
+							e -> e.transactional(true))
+					.channel(c -> c.queue("persistResults"));
+		}
+
+		@Bean
+		public IntegrationFlow retrievingGatewayFlow(EntityManagerFactory entityManagerFactory) {
+			return f -> f
+					.handle(Jpa.retrievingGateway(entityManagerFactory)
+							.jpaQuery("from Student s where s.id = :id")
+							.expectSingleResult(true)
+							.parameterExpression("id", "payload"))
+					.channel(c -> c.queue("retrieveResults"));
+		}
+
+	}
+
+}