From 4a85849bcc4fed5dcc3842f17afc1cebc8b785e0 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 2 Aug 2018 17:41:25 -0400 Subject: [PATCH] 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 --- .../integration/jdbc/JdbcMessageHandler.java | 164 +++++++++++++----- .../JdbcMessageHandlerIntegrationTests.java | 44 ++++- src/reference/asciidoc/jdbc.adoc | 10 +- src/reference/asciidoc/whats-new.adoc | 2 + 4 files changed, 173 insertions(+), 47 deletions(-) diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java index b066df73a8..5e826f51fc 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java +++ b/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,22 +18,26 @@ package org.springframework.integration.jdbc; 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; @@ -41,47 +45,59 @@ 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. * * 
  * INSERT INTO FOOS (MESSAGE_ID, PAYLOAD) VALUES (:headers[id], :payload)
  *
* + *

+ * 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. + *

+ * When a {@link #preparedStatementSetter} is configured, it is applied for each item in the appropriate + * {@link JdbcOperations#batchUpdate(String, BatchPreparedStatementSetter)} function. + *

+ * 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. business.id) * * @author Dave Syer * @author Artem Bilan + * * @since 2.0 */ public class JdbcMessageHandler extends AbstractMessageHandler { private final ResultSetExtractor>> generatedKeysResultSetExtractor = - new RowMapperResultSetExtractor>(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 class JdbcMessageHandler extends AbstractMessageHandler { /** * 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 class JdbcMessageHandler extends AbstractMessageHandler { 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,7 +169,7 @@ public class JdbcMessageHandler extends AbstractMessageHandler { * Executes the update, passing the message into the {@link SqlParameterSourceFactory}. */ @Override - protected void handleMessageInternal(Message message) throws Exception { + protected void handleMessageInternal(Message message) { List> keys = executeUpdateQuery(message, this.keysGenerated); if (!keys.isEmpty() && logger.isDebugEnabled()) { logger.debug("Generated keys: " + keys); @@ -157,49 +177,103 @@ public class JdbcMessageHandler extends AbstractMessageHandler { } protected List> 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>>) ps -> { - JdbcMessageHandler.this.preparedStatementSetter.setValues(ps, message); - ps.executeUpdate(); - ResultSet keys = ps.getGeneratedKeys(); // NOSONAR closed in JdbcUtils - if (keys != null) { - try { + 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 JdbcMessageHandler.this.generatedKeysResultSetExtractor.extractData(keys); - } - finally { - JdbcUtils.closeResultSet(keys); - } - } - return new LinkedList>(); - }); + 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> messageStream = + StreamSupport.stream(((Iterable) message.getPayload()).spliterator(), false) + .map(payload -> new Message() { + + @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 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 map = new LinkedCaseInsensitiveMap<>(); + map.put("UPDATED", updated); + return Collections.singletonList(map); } - LinkedCaseInsensitiveMap map = new LinkedCaseInsensitiveMap(); - map.put("UPDATED", updated); - return Collections.singletonList(map); } } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageHandlerIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageHandlerIntegrationTests.java index 8fea0568de..c54d8f50af 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageHandlerIntegrationTests.java +++ b/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 @@ package org.springframework.integration.jdbc; 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 class JdbcMessageHandlerIntegrationTests { 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> message = new GenericMessage<>(Arrays.asList("foo1", "foo2", "foo3")); + handler.handleMessage(message); + + List> 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 class JdbcMessageHandlerIntegrationTests { 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> message = new GenericMessage<>(Arrays.asList("foo1", "foo2", "foo3")); + handler.handleMessage(message); + + List> 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, diff --git a/src/reference/asciidoc/jdbc.adoc b/src/reference/asciidoc/jdbc.adoc index 87187d6e26..4e21bbf88b 100644 --- a/src/reference/asciidoc/jdbc.adoc +++ b/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 `` 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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 41ba8d87eb..d28ee8ad50 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -109,6 +109,8 @@ See <> 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 <> for more information. [[x5.1-ftp-sftp]]