From 0798aad75c394c794cd21c69578dd23199fb803a Mon Sep 17 00:00:00 2001 From: Jonas Partner Date: Tue, 9 Feb 2010 21:33:45 +0000 Subject: [PATCH] start of JDBC inbound channel adapter INT-790 --- .../jdbc/JdbcPollingChannelAdapter.java | 145 ++++++++++++++++++ ...cPollingChannelAdapterIntegrationTest.java | 117 ++++++++++++++ .../src/test/resources/log4j.properties | 10 ++ .../pollingChannelAdapterIntegrationTest.sql | 1 + 4 files changed, 273 insertions(+) create mode 100644 org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java create mode 100644 org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java create mode 100644 org.springframework.integration.jdbc/src/test/resources/log4j.properties create mode 100644 org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/pollingChannelAdapterIntegrationTest.sql diff --git a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java new file mode 100644 index 0000000000..55a653c4ba --- /dev/null +++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java @@ -0,0 +1,145 @@ +/* + * Copyright 2002-2008 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.jdbc; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageSource; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.core.namedparam.SqlParameterSource; +import org.springframework.jdbc.core.simple.SimpleJdbcOperations; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.DefaultTransactionDefinition; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; + +import javax.sql.DataSource; +import java.util.List; + +/** + * A polling channel adapter that creates messages from the payload returned by executing a select query + * Optionally an update can be executed after the select in order to update processed rows + * + * @author Jonas Partner + */ +public class JdbcPollingChannelAdapter implements MessageSource, InitializingBean { + + private final SimpleJdbcOperations jdbcOperations; + + private final String selectQuery; + + private volatile RowMapper rowMapper; + + private volatile String updateQuery; + + private volatile SqlParameterSource sqlQueryParameterSource; + + private volatile TransactionDefinition transactionDefinition; + + private volatile TransactionTemplate transactionTemplate; + + private volatile PlatformTransactionManager platformTransactionManager; + + public JdbcPollingChannelAdapter(DataSource dataSource, String selectQuery) { + this.jdbcOperations = new SimpleJdbcTemplate(dataSource); + this.selectQuery = selectQuery; + } + + public JdbcPollingChannelAdapter(SimpleJdbcOperations jdbcOperations, String selectQuery) { + this.jdbcOperations = jdbcOperations; + this.selectQuery = selectQuery; + } + + public void setTransactionDefinition(TransactionDefinition transactionDefinition) { + this.transactionDefinition = transactionDefinition; + } + + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.platformTransactionManager = platformTransactionManager; + } + + public void setRowMapper(RowMapper rowMapper) { + this.rowMapper = rowMapper; + } + + public void setUpdateQuery(String updateQuery) { + this.updateQuery = updateQuery; + } + + + public void afterPropertiesSet() throws Exception { + if (this.transactionDefinition == null) { + this.transactionDefinition = new DefaultTransactionDefinition(); + } + + if (this.platformTransactionManager != null) { + this.transactionTemplate = new TransactionTemplate(this.platformTransactionManager, this.transactionDefinition); + } + } + + public Message receive() { + Object payload = null; + if (this.transactionTemplate != null){ + payload = this.transactionTemplate.execute(new TransactionCallback(){ + public Object doInTransaction(TransactionStatus status) { + return pollAndUpdate(); + } + }); + } else { + payload = pollAndUpdate(); + } + return MessageBuilder.withPayload(payload).build(); + } + + protected Object pollAndUpdate(){ + List payload; + if (this.rowMapper != null) { + payload = pollWithRowMapper(); + } else { + payload = this.jdbcOperations.queryForList(this.selectQuery, this.sqlQueryParameterSource); + } + + return payload; + + } + + protected List pollWithRowMapper(){ + List payload = null; + if(this.sqlQueryParameterSource != null){ + payload = this.jdbcOperations.query(this.selectQuery, this.rowMapper, this.sqlQueryParameterSource); + } else { + payload = this.jdbcOperations.query(this.selectQuery, this.rowMapper); + } + return payload; + } + + protected Object pollForListOfMap(){ + List payload = null; + if(this.sqlQueryParameterSource != null){ + payload = this.jdbcOperations.queryForList(this.selectQuery, this.sqlQueryParameterSource); + } else { + payload = this.jdbcOperations.queryForList(this.selectQuery); + } + return payload; + } + + +} diff --git a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java new file mode 100644 index 0000000000..43970903ce --- /dev/null +++ b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTest.java @@ -0,0 +1,117 @@ +package org.springframework.integration.jdbc; + +import static org.junit.Assert.*; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.List; +import java.util.Map; + +import org.junit.After; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.springframework.integration.core.Message; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; + +/** + * @author Jonas Partner + */ +public class JdbcPollingChannelAdapterIntegrationTest { + + EmbeddedDatabase embeddedDatabase; + + SimpleJdbcTemplate jdbcTemplate; + + @Before + public void setUp(){ + EmbeddedDatabaseBuilder builder = new EmbeddedDatabaseBuilder(); + builder.setType(EmbeddedDatabaseType.DERBY).addScript("classpath:org/springframework/integration/jdbc/pollingChannelAdapterIntegrationTest.sql"); + this.embeddedDatabase = builder.build(); + this.jdbcTemplate = new SimpleJdbcTemplate(this.embeddedDatabase); + } + + + @After + public void tearDown(){ + this.embeddedDatabase.shutdown(); + } + + @Test + public void testSimplePollForListOfMapsNoUpdate(){ + JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(this.embeddedDatabase, "select * from item"); + this.jdbcTemplate.update("insert into item values(1,2)") ; + Message message = adapter.receive(); + Object payload = message.getPayload(); + assertTrue("Wrong payload type", payload instanceof List); + List rows = (List)payload; + assertEquals("Wrong number of elements" , 1, rows.size()); + assertTrue("Returned row not a map", rows.get(0) instanceof Map); + Map row = (Map)rows.get(0); + assertEquals("Wrong id", 1, row.get("id")); + assertEquals("Wrong status", 2, row.get("status")); + + } + + + @Test + public void testSimplePollForListWithRowMapperNoUpdate(){ + JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(this.embeddedDatabase, "select * from item"); + adapter.setRowMapper(new ItemRowMapper()); + this.jdbcTemplate.update("insert into item values(1,2)") ; + Message message = adapter.receive(); + Object payload = message.getPayload(); + List rows = (List)payload; + assertEquals("Wrong number of elements" , 1, rows.size()); + assertTrue("Wrong payload type", rows.get(0) instanceof Item); + Item item = (Item)rows.get(0); + assertEquals("Wrong id", 1, item.getId()); + assertEquals("Wrong status", 2, item.getStatus()); + + } + + private static class Item{ + + + private int id; + + private int status; + + public int getId() { + return id; + } + + public void setId(int id) { + this.id = id; + } + + public int getStatus() { + return status; + } + + public void setStatus(int status) { + this.status = status; + } + } + + private static class ItemRowMapper implements RowMapper{ + + + public Item mapRow(ResultSet rs, int rowNum) throws SQLException { + Item item = new Item(); + item.setId(rs.getInt(1)); + item.setStatus(rs.getInt(2)); + return item; + } + + + + + + } + +} diff --git a/org.springframework.integration.jdbc/src/test/resources/log4j.properties b/org.springframework.integration.jdbc/src/test/resources/log4j.properties new file mode 100644 index 0000000000..ba5b291f5c --- /dev/null +++ b/org.springframework.integration.jdbc/src/test/resources/log4j.properties @@ -0,0 +1,10 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n + + +log4j.category.org.springframework=DEBUG +log4j.category.org.springframework.integration=INFO +log4j.category.org.springframework.integration.jdbc=DEBUG \ No newline at end of file diff --git a/org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/pollingChannelAdapterIntegrationTest.sql b/org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/pollingChannelAdapterIntegrationTest.sql new file mode 100644 index 0000000000..1f0246d3b8 --- /dev/null +++ b/org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/pollingChannelAdapterIntegrationTest.sql @@ -0,0 +1 @@ +create table item(id int,status int); \ No newline at end of file