INT-3364: Add batchUpdate into JdbcMessageHandler (#2534)

* INT-3364: Add batchUpdate into JdbcMessageHandler

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

* * Optimize items mapping with an internal `Message` implementation
* Polishing Docs and Javadocs

Author: artembilan

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -18,70 +18,86 @@
 
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 import javax.sql.DataSource;
 
 import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.jdbc.core.BatchPreparedStatementSetter;
 import org.springframework.jdbc.core.ColumnMapRowMapper;
 import org.springframework.jdbc.core.JdbcOperations;
-import org.springframework.jdbc.core.PreparedStatementCallback;
 import org.springframework.jdbc.core.PreparedStatementCreator;
 import org.springframework.jdbc.core.ResultSetExtractor;
 import org.springframework.jdbc.core.RowMapperResultSetExtractor;
-import org.springframework.jdbc.core.namedparam.EmptySqlParameterSource;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.jdbc.core.namedparam.SqlParameterSource;
 import org.springframework.jdbc.support.GeneratedKeyHolder;
 import org.springframework.jdbc.support.JdbcUtils;
 import org.springframework.jdbc.support.KeyHolder;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
 import org.springframework.util.LinkedCaseInsensitiveMap;
 
 /**
  * A message handler that executes an SQL update. Dynamic query parameters are supported through the
  * {@link SqlParameterSourceFactory} abstraction, the default implementation of which wraps the message so that its bean
- * properties can be referred to by name in the query string E.g.
+ * properties can be referred to by name in the query string, e.g.
  *
  * <pre class="code">
  * INSERT INTO FOOS (MESSAGE_ID, PAYLOAD) VALUES (:headers[id], :payload)
  * </pre>
  *
+ * <p>
+ * When a message payload is an instance of {@link Iterable}, a
+ * {@link NamedParameterJdbcOperations#batchUpdate(String, SqlParameterSource[])} is performed, where each
+ * {@link SqlParameterSource} instance is based on items wrapped into an internal {@link Message} implementation with
+ * headers from the request message.
+ * <p>
+ * When a {@link #preparedStatementSetter} is configured, it is applied for each item in the appropriate
+ * {@link JdbcOperations#batchUpdate(String, BatchPreparedStatementSetter)} function.
+ * <p>
+ * NOTE: The batch update is not supported when {@link #keysGenerated} is in use.
+ *
  * N.B. do not use quotes to escape the header keys. The default SQL parameter source (from Spring JDBC) can also handle
  * headers with dotted names (e.g. <code>business.id</code>)
  *
  * @author Dave Syer
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class JdbcMessageHandler extends AbstractMessageHandler {
 
 	private final ResultSetExtractor<List<Map<String, Object>>> generatedKeysResultSetExtractor =
-			new RowMapperResultSetExtractor<Map<String, Object>>(new ColumnMapRowMapper(), 1);
+			new RowMapperResultSetExtractor<>(new ColumnMapRowMapper(), 1);
 
 	private final NamedParameterJdbcOperations jdbcOperations;
 
-	private final PreparedStatementCreator generatedKeysStatementCreator = con ->
-			con.prepareStatement(JdbcMessageHandler.this.updateSql, Statement.RETURN_GENERATED_KEYS);
+	private final PreparedStatementCreator generatedKeysStatementCreator =
+			con -> con.prepareStatement(JdbcMessageHandler.this.updateSql, Statement.RETURN_GENERATED_KEYS);
 
-	private volatile String updateSql;
+	private String updateSql;
 
-	private volatile SqlParameterSourceFactory sqlParameterSourceFactory;
+	private SqlParameterSourceFactory sqlParameterSourceFactory;
 
-	private volatile boolean keysGenerated;
+	private boolean keysGenerated;
 
 	private MessagePreparedStatementSetter preparedStatementSetter;
 
 	/**
 	 * Constructor taking {@link DataSource} from which the DB Connection can be obtained and the select query to
 	 * execute to retrieve new rows.
-	 *
 	 * @param dataSource Must not be null
 	 * @param updateSql query to execute
 	 */
@@ -93,13 +109,12 @@ public JdbcMessageHandler(DataSource dataSource, String updateSql) {
 	/**
 	 * Constructor taking {@link JdbcOperations} instance to use for query execution and the select query to execute to
 	 * retrieve new rows.
-	 *
 	 * @param jdbcOperations instance to use for query execution
 	 * @param updateSql query to execute
 	 */
 	public JdbcMessageHandler(JdbcOperations jdbcOperations, String updateSql) {
 		this.jdbcOperations = new NamedParameterJdbcTemplate(jdbcOperations);
-		this.updateSql = updateSql;
+		setUpdateSql(updateSql);
 	}
 
 	/**
@@ -111,7 +126,12 @@ public void setKeysGenerated(boolean keysGenerated) {
 		this.keysGenerated = keysGenerated;
 	}
 
-	public void setUpdateSql(String updateSql) {
+	/**
+	 * Configure an SQL statement to perform an UPDATE on the target database.
+	 * @param updateSql the SQL statement to perform.
+	 */
+	public final void setUpdateSql(String updateSql) {
+		Assert.hasText(updateSql, "'updateSql' must not be empty.");
 		this.updateSql = updateSql;
 	}
 
@@ -149,57 +169,111 @@ protected void onInit() throws Exception {
 	 * Executes the update, passing the message into the {@link SqlParameterSourceFactory}.
 	 */
 	@Override
-	protected void handleMessageInternal(Message<?> message) throws Exception {
+	protected void handleMessageInternal(Message<?> message) {
 		List<? extends Map<String, Object>> keys = executeUpdateQuery(message, this.keysGenerated);
 		if (!keys.isEmpty() && logger.isDebugEnabled()) {
 			logger.debug("Generated keys: " + keys);
 		}
 	}
 
 	protected List<? extends Map<String, Object>> executeUpdateQuery(final Message<?> message, boolean keysGenerated) {
-		SqlParameterSource updateParameterSource = EmptySqlParameterSource.INSTANCE;
-		if (this.preparedStatementSetter == null) {
-			if (this.sqlParameterSourceFactory != null) {
-				updateParameterSource = this.sqlParameterSourceFactory.createParameterSource(message);
-			}
-		}
 		if (keysGenerated) {
 			if (this.preparedStatementSetter != null) {
-				return this.jdbcOperations.getJdbcOperations().execute(this.generatedKeysStatementCreator,
-						(PreparedStatementCallback<List<Map<String, Object>>>) ps -> {
-							JdbcMessageHandler.this.preparedStatementSetter.setValues(ps, message);
-							ps.executeUpdate();
-							ResultSet keys = ps.getGeneratedKeys(); // NOSONAR closed in JdbcUtils
-							if (keys != null) {
-								try {
-
-									return JdbcMessageHandler.this.generatedKeysResultSetExtractor.extractData(keys);
-								}
-								finally {
-									JdbcUtils.closeResultSet(keys);
-								}
-							}
-							return new LinkedList<Map<String, Object>>();
-						});
+				return this.jdbcOperations.getJdbcOperations()
+						.execute(this.generatedKeysStatementCreator,
+								ps -> {
+									this.preparedStatementSetter.setValues(ps, message);
+									ps.executeUpdate();
+									ResultSet keys = ps.getGeneratedKeys(); // NOSONAR closed in JdbcUtils
+									if (keys != null) {
+										try {
+
+											return this.generatedKeysResultSetExtractor.extractData(keys);
+										}
+										finally {
+											JdbcUtils.closeResultSet(keys);
+										}
+									}
+									return new LinkedList<>();
+								});
 			}
 			else {
 				KeyHolder keyHolder = new GeneratedKeyHolder();
-				this.jdbcOperations.update(this.updateSql, updateParameterSource, keyHolder);
+				this.jdbcOperations.update(this.updateSql,
+						this.sqlParameterSourceFactory.createParameterSource(message), keyHolder);
 				return keyHolder.getKeyList();
 			}
 		}
 		else {
-			int updated;
-			if (this.preparedStatementSetter != null) {
-				updated = this.jdbcOperations.getJdbcOperations().update(this.updateSql,
-						ps -> JdbcMessageHandler.this.preparedStatementSetter.setValues(ps, message));
+			if (message.getPayload() instanceof Iterable) {
+				Stream<? extends Message<?>> messageStream =
+						StreamSupport.stream(((Iterable<?>) message.getPayload()).spliterator(), false)
+								.map(payload -> new Message<Object>() {
+
+									@Override
+									public Object getPayload() {
+										return payload;
+									}
+
+									@Override
+									public MessageHeaders getHeaders() {
+										return message.getHeaders();
+									}
+
+								});
+
+				int[] updates;
+
+				if (this.preparedStatementSetter != null) {
+					Message<?>[] messages = messageStream.toArray(Message<?>[]::new);
+
+					updates = this.jdbcOperations.getJdbcOperations()
+							.batchUpdate(this.updateSql, new BatchPreparedStatementSetter() {
+
+								@Override
+								public void setValues(PreparedStatement ps, int i) throws SQLException {
+									JdbcMessageHandler.this.preparedStatementSetter.setValues(ps, messages[i]);
+								}
+
+								@Override
+								public int getBatchSize() {
+									return messages.length;
+								}
+
+							});
+				}
+				else {
+					SqlParameterSource[] sqlParameterSources =
+							messageStream.map(this.sqlParameterSourceFactory::createParameterSource)
+									.toArray(SqlParameterSource[]::new);
+
+					updates = this.jdbcOperations.batchUpdate(this.updateSql, sqlParameterSources);
+				}
+
+				return Arrays.stream(updates)
+						.mapToObj(updated -> {
+							Map<String, Object> map = new LinkedCaseInsensitiveMap<>();
+							map.put("UPDATED", updated);
+							return map;
+						})
+						.collect(Collectors.toList());
 			}
 			else {
-				updated = this.jdbcOperations.update(this.updateSql, updateParameterSource);
+				int updated;
+
+				if (this.preparedStatementSetter != null) {
+					updated = this.jdbcOperations.getJdbcOperations()
+							.update(this.updateSql, ps -> this.preparedStatementSetter.setValues(ps, message));
+				}
+				else {
+					updated = this.jdbcOperations.update(this.updateSql,
+							this.sqlParameterSourceFactory.createParameterSource(message));
+				}
+
+				LinkedCaseInsensitiveMap<Object> map = new LinkedCaseInsensitiveMap<>();
+				map.put("UPDATED", updated);
+				return Collections.singletonList(map);
 			}
-			LinkedCaseInsensitiveMap<Object> map = new LinkedCaseInsensitiveMap<Object>();
-			map.put("UPDATED", updated);
-			return Collections.singletonList(map);
 		}
 	}
 

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageHandlerIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-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.
@@ -19,6 +19,8 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -88,6 +90,24 @@ public void testSimpleDynamicInsert() {
 		assertEquals("Wrong name", "foo", map.get("NAME"));
 	}
 
+	@Test
+	public void testInsertBatch() {
+		JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate,
+				"insert into foos (id, status, name) values (:payload, 0, :payload)");
+		handler.afterPropertiesSet();
+
+		Message<List<String>> message = new GenericMessage<>(Arrays.asList("foo1", "foo2", "foo3"));
+		handler.handleMessage(message);
+
+		List<Map<String, Object>> foos = jdbcTemplate.queryForList("SELECT * FROM FOOS ORDER BY id");
+
+		assertEquals(3, foos.size());
+
+		assertEquals("foo1", foos.get(0).get("NAME"));
+		assertEquals("foo2", foos.get(1).get("NAME"));
+		assertEquals("foo3", foos.get(2).get("NAME"));
+	}
+
 	@Test
 	public void testInsertWithMessagePreparedStatementSetter() {
 		JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate,
@@ -105,6 +125,28 @@ public void testInsertWithMessagePreparedStatementSetter() {
 		assertTrue(setterInvoked.get());
 	}
 
+	@Test
+	public void testInsertBatchWithMessagePreparedStatementSetter() {
+		JdbcMessageHandler handler =
+				new JdbcMessageHandler(jdbcTemplate, "insert into foos (id, status, name) values (?, 0, ?)");
+		handler.setPreparedStatementSetter((ps, requestMessage) -> {
+			ps.setObject(1, requestMessage.getPayload());
+			ps.setObject(2, requestMessage.getPayload());
+		});
+		handler.afterPropertiesSet();
+
+		Message<List<String>> message = new GenericMessage<>(Arrays.asList("foo1", "foo2", "foo3"));
+		handler.handleMessage(message);
+
+		List<Map<String, Object>> foos = jdbcTemplate.queryForList("SELECT * FROM FOOS ORDER BY id");
+
+		assertEquals(3, foos.size());
+
+		assertEquals("foo1", foos.get(0).get("NAME"));
+		assertEquals("foo2", foos.get(1).get("NAME"));
+		assertEquals("foo3", foos.get(2).get("NAME"));
+	}
+
 	@Test
 	public void testIdHeaderDynamicInsert() {
 		JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate,

src/reference/asciidoc/jdbc.adoc

@@ -261,7 +261,15 @@ public MessageHandler jdbcMessageHandler(DataSource dataSource) {
 ====
 
 From the XML configuration perspective, the `prepared-statement-setter` attribute is available on the `<int-jdbc:outbound-channel-adapter>` component.
-It lets you specify a  `MessagePreparedStatementSetter` bean reference.
+It lets you specify a `MessagePreparedStatementSetter` bean reference.
+
+==== Batch Update
+
+Starting with version 5.1, the `JdbcMessageHandler` performs a `JdbcOperations.batchUpdate()` if the payload of the request message is an `Iterable` instance.
+Each element of the `Iterable` is wrapped to a `Message` with the headers from the request message.
+In the case of regular `SqlParameterSourceFactory`-based configuration these messages are used to build an `SqlParameterSource[]` for an argument used in the mentioned `JdbcOperations.batchUpdate()` function.
+When a `MessagePreparedStatementSetter` configuration is applied, a `BatchPreparedStatementSetter` variant is used to iterate over those messages for each item and the provided `MessagePreparedStatementSetter` is called against them.
+The batch update is not supported when `keysGenerated` mode is selected.
 
 [[jdbc-outbound-gateway]]
 === Outbound Gateway

src/reference/asciidoc/whats-new.adoc

@@ -109,6 +109,8 @@ See <<amqp-content-type>> for more information.
 
 A confusing `max-rows-per-poll` property on the JDBC Inbound Channel Adapter and JDBC Outbound Gateway has been deprecated in favor of the newly introduced `max-rows` property.
 
+The `JdbcMessageHandler` supports now a `batchUpdate` functionality when the payload of the request message is an instance of an `Iterable` type.
+
 See <<jdbc>> for more information.
 
 [[x5.1-ftp-sftp]]