The inbound adapter accepts a regular Spring Integration poller as a child element. Consequently, the frequency of the polling can be controlled (among other uses). An important feature of the poller for JDBC usage is the option to wrap the poll operation in a transaction, as the following example shows:

<int-jdbc:inbound-channel-adapter query="..."
        channel="target" data-source="dataSource" update="...">
    <int:poller fixed-rate="1000">
        <int:transactional/>
    </int:poller>
</int-jdbc:inbound-channel-adapter>

	Note
	If you do not explicitly specify a poller, a default value is used. As is normal with Spring Integration, it can be defined as a top-level bean).

In the preceding example, the database is polled every 1000 milliseconds (or once a second), and the update and select queries are both executed in the same transaction. The transaction manager configuration is not shown. However, as long as it is aware of the data source, the poll is transactional. A common use case is for the downstream channels to be direct channels (the default), so that the endpoints are invoked in the same thread and, hence, the same transaction. That way, if any of them fail, the transaction rolls back and the input data is reverted to its original state.

The JDBC inbound channel adapter defines an attribute called max-rows. When you specify the adapter’s poller, you can also define a property called max-messages-per-poll. While these two attributes look similar, their meaning is quite different.

max-messages-per-poll specifies the number of times the query is executed per polling interval, whereas max-rows specifies the number of rows returned for each execution.

Under normal circumstances, you would likely not want to set the poller’s max-messages-per-poll property when you use the JDBC inbound channel adapter. Its default value is 1, which means that the JDBC inbound channel adapter’s receive() method is executed exactly once for each poll interval.

Setting the max-messages-per-poll attribute to a larger value means that the query is executed that many times back to back. For more information regarding the max-messages-per-poll attribute, see Section 6.3.1, “Configuring An Inbound Channel Adapter”.

In contrast, the max-rows attribute, if greater than 0, specifies the maximum number of rows to be used from the query result set created by the receive() method. If the attribute is set to 0, all rows are included in the resulting message. The attribute defaults to 0.

	Note
	It is recommended to use result set limiting via vendor-specific query options, for example MySQL LIMIT or SQL Server TOP or Oracle’s ROWNUM. See the particular vendor documentation for more information.

A common requirement for most JDBC channel adapters is to pass parameters as part of SQL queries or stored procedures or functions. As mentioned earlier, these parameters are by default bean property expressions, not SpEL expressions. However, if you need to pass SpEL expression as parameters, you must explicitly inject a SqlParameterSourceFactory.

The following example uses a ExpressionEvaluatingSqlParameterSourceFactory to achieve that requirement:

<jdbc:outbound-channel-adapter data-source="dataSource" channel="input"
    query="insert into MESSAGES (MESSAGE_ID,PAYLOAD,CREATED_DATE)     \
    values (:id, :payload, :createdDate)"
    sql-parameter-source-factory="spelSource"/>

<bean id="spelSource"
      class="o.s.integration.jdbc.ExpressionEvaluatingSqlParameterSourceFactory">
    <property name="parameterExpressions">
        <map>
            <entry key="id"          value="headers['id'].toString()"/>
            <entry key="createdDate" value="new java.util.Date()"/>
            <entry key="payload"     value="payload"/>
        </map>
    </property>
</bean>

For further information, see Section 21.5.5, “Defining Parameter Sources”.

Sometimes, the flexibility and loose-coupling of SqlParameterSourceFactory does not do what we need for the target PreparedStatement or we need to do some low-level JDBC work. The Spring JDBC module provides APIs to configure the execution environment (such as ConnectionCallback or PreparedStatementCreator) and manipulate parameter values (such as SqlParameterSource). It can even access APIs for low-level operations, such as StatementCallback.

Starting with Spring Integration 4.2, MessagePreparedStatementSetter allows the specification of parameters on the PreparedStatement manually, in the requestMessage context. This class plays exactly the same role as PreparedStatementSetter in the standard Spring JDBC API. Actually, it is invoked directly from an inline PreparedStatementSetter implementation when the JdbcMessageHandler invokes execute on the JdbcTemplate.

This functional interface option is mutually exclusive with sqlParameterSourceFactory and can be used as a more powerful alternative to populate parameters of the PreparedStatement from the requestMessage. For example, it is useful when we need to store File data to the DataBase BLOB column in a streaming manner. The following example shows how to do so:

@Bean
@ServiceActivator(inputChannel = "storeFileChannel")
public MessageHandler jdbcMessageHandler(DataSource dataSource) {
    JdbcMessageHandler jdbcMessageHandler = new JdbcMessageHandler(dataSource,
            "INSERT INTO imagedb (image_name, content, description) VALUES (?, ?, ?)");
    jdbcMessageHandler.setPreparedStatementSetter((ps, m) -> {
        ps.setString(1, m.getHeaders().get(FileHeaders.FILENAME));
        try (FileInputStream inputStream = new FileInputStream((File) m.getPayload()); ) {
            ps.setBlob(2, inputStream);
        }
        catch (Exception e) {
            throw new MessageHandlingException(m, e);
        }
        ps.setClob(3, new StringReader(m.getHeaders().get("description", String.class)));
    });
    return jdbcMessageHandler;
}

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.

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.

Before starting to use JDBC message store components, you should provision a target database with the appropriate objects.

Spring Integration ships with some sample scripts that can be used to initialize a database. In the spring-integration-jdbc JAR file, you can find scripts in the org.springframework.integration.jdbc package. It provides an example create and an example drop script for a range of common database platforms. A common way to use these scripts is to reference them in a Spring JDBC data source initializer. Note that the scripts are provided as samples and as specifications of the the required table and column names. You may find that you need to enhance them for production use (for, example, by adding index declarations).

The JDBC module provides an implementation of the Spring Integration MessageStore (important in the claim check pattern) and MessageGroupStore (important in stateful patterns such as an aggregator) backed by a database. Both interfaces are implemented by the JdbcMessageStore, and there is support for configuring store instances in XML, as the following example shows:

<int-jdbc:message-store id="messageStore" data-source="dataSource"/>

You can specify a JdbcTemplate instead of a DataSource.

The following example shows some other optional attributes:

<int-jdbc:message-store id="messageStore" data-source="dataSource"
    lob-handler="lobHandler" table-prefix="MY_INT_"/>

In the preceding example, we have specified a LobHandler for dealing with messages as large objects (which is often necessary for Oracle) and a prefix for the table names in the queries generated by the store. The table name prefix defaults to INT_.

If you intend to backing message channels with JDBC, we recommend using the JdbcChannelMessageStore implementation. It works only in conjunction with Message Channels.

Custom Message Insertion

Since version 5.0, by overloading the ChannelMessageStorePreparedStatementSetter class, you can provide a custom implementation for message insertion in the JdbcChannelMessageStore. You can use it to set different columns or change the table structure or serialization strategy. For example, instead of default serialization to byte[], you can store its structure as a JSON string.

The following example uses the default implementation of setValues to store common columns and overrides the behavior to store the message payload as a varchar:

public class JsonPreparedStatementSetter extends ChannelMessageStorePreparedStatementSetter {

    public JsonPreparedStatementSetter() {
        super();
    }

    @Override
    public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage,
        Object groupId, String region, 	boolean priorityEnabled) throws SQLException {
        // Populate common columns
        super.setValues(preparedStatement, requestMessage, groupId, region, priorityEnabled);
        // Store message payload as varchar
        preparedStatement.setString(6, requestMessage.getPayload().toString());
    }
}

	Important
	Generally, we do not recommend using a relational database for queuing. Instead, if possible, consider using either JMS- or AMQP-backed channels instead. For further reference, see the following resources: 5 subtle ways you’re using MySQL as a queue, and why it’ll bite you. The Database As Queue Anti-Pattern.

Concurrent Polling

When polling a message channel, you have the option to configure the associated Poller with a TaskExecutor reference.

Important

Keep in mind, though, that if you use a JDBC backed message channel and you plan to poll the channel and consequently the message store transactionally with multiple threads, you should ensure that you use a relational database that supports Multiversion Concurrency Control (MVCC). Otherwise, locking may be an issue and the performance, when using multiple threads, may not materialize as expected. For example, Apache Derby is problematic in that regard. To achieve better JDBC queue throughput and avoid issues when different threads may poll the same Message from the queue, it is important to set the usingIdCache property of JdbcChannelMessageStore to true when using databases that do not support MVCC. The following example shows how to do so: <bean id="queryProvider" class="o.s.i.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider"/> <int:transaction-synchronization-factory id="syncFactory"> <int:after-commit expression="@store.removeFromIdCache(headers.id.toString())" /> <int:after-rollback expression="@store.removeFromIdCache(headers.id.toString())"/> </int:transaction-synchronization-factory> <task:executor id="pool" pool-size="10" queue-capacity="10" rejection-policy="CALLER_RUNS" /> <bean id="store" class="o.s.i.jdbc.store.JdbcChannelMessageStore"> <property name="dataSource" ref="dataSource"/> <property name="channelMessageStoreQueryProvider" ref="queryProvider"/> <property name="region" value="TX_TIMEOUT"/> <property name="usingIdCache" value="true"/> </bean> <int:channel id="inputChannel"> <int:queue message-store="store"/> </int:channel> <int:bridge input-channel="inputChannel" output-channel="outputChannel"> <int:poller fixed-delay="500" receive-timeout="500" max-messages-per-poll="1" task-executor="pool"> <int:transactional propagation="REQUIRED" synchronization-factory="syncFactory" isolation="READ_COMMITTED" transaction-manager="transactionManager" /> </int:poller> </int:bridge> <int:channel id="outputChannel" />

Priority Channel

Starting with version 4.0, JdbcChannelMessageStore implements PriorityCapableChannelMessageStore and provides the priorityEnabled option, letting it be used as a message-store reference for priority-queue instances. For this purpose, the INT_CHANNEL_MESSAGE table has a MESSAGE_PRIORITY column to store the value of PRIORITY message headers. In addition, a new MESSAGE_SEQUENCE column lets us achieve a robust first-in-first-out (FIFO) polling mechanism, even when multiple messages are stored with the same priority in the same millisecond. Messages are polled (selected) from the database with order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE.

Note

We do not recommend using the same JdbcChannelMessageStore bean for priority and non-priority queue channels, because the priorityEnabled option applies to the entire store and proper FIFO queue semantics are not retained for the queue channel. However, the same INT_CHANNEL_MESSAGE table (and even region) can be used for both JdbcChannelMessageStore types. To configure that scenario, you can extend one message store bean from the other, as the following example shows:

<bean id="channelStore" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
    <property name="dataSource" ref="dataSource"/>
    <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
</bean>

<int:channel id="queueChannel">
    <int:queue message-store="channelStore"/>
</int:channel>

<bean id="priorityStore" parent="channelStore">
    <property name="priorityEnabled" value="true"/>
</bean>

<int:channel id="priorityChannel">
    <int:priority-queue message-store="priorityStore"/>
</int:channel>

It is common to use a JdbcMessageStore as a global store for a group of applications or nodes in the same application. To provide some protection against name clashes and to give control over the database meta-data configuration, the message store lets the tables be partitioned in two ways. One way is to use separate table names, by changing the prefix (as described earlier). The other way is to specify a region name for partitioning data within a single table. An important use case for the second approach is when the MessageStore is managing persistent queues that back a Spring Integration Message Channel. The message data for a persistent channel is keyed in the store on the channel name. Consequently, if the channel names are not globally unique, the channels can pick up data that is not intended for them. To avoid this danger, you can use the message store region to keep data separate for different physical channels that have the same logical name.

In order to enable calls to stored procedures and stored functions, the stored procedure components use the org.springframework.jdbc.core.simple.SimpleJdbcCall class. Consequently, the following databases are fully supported for executing stored procedures:

If you want to execute stored functions instead, the following databases are fully supported:

	Note
	Even though your particular database may not be fully supported, chances are that you can use the stored procedure Spring Integration components quite successfully anyway, provided your RDBMS supports stored procedures or stored functions. As a matter of fact, some of the provided integration tests use the H2 database. Nevertheless, it is very important to thoroughly test those usage scenarios.

The stored procedure components provide full XML Namespace support, and configuring the components is similar as for the general purpose JDBC components discussed earlier.

All stored procedure components share certain configuration parameters:

The stored procedure components share a common set of child elements that you can use to define and pass parameters to stored procedures or stored functions. The following elements are available:

Parameter sources govern the techniques of retrieving and mapping the Spring Integration message properties to the relevant stored procedure input parameters.

The stored procedure components follow certain rules. By default, the bean properties of the Message payload are used as a source for the stored procedure’s input parameters. In that case, a BeanPropertySqlParameterSourceFactory is used. This may suffice for basic use cases. The next example illustrates that default behavior.

Important

For the "automatic" lookup of bean properties by using the BeanPropertySqlParameterSourceFactory to work, your bean properties must be defined in lower case. This is due to the fact that in org.springframework.jdbc.core.metadata.CallMetaDataContext (the Java method is matchInParameterValuesWithCallParameters()), the retrieved stored procedure parameter declarations are converted to lower case. As a result, if you have camel-case bean properties (such as lastName), the lookup fails. In that case, provide an explicit ProcedureParameter.

Suppose we have a payload that consists of a simple bean with the following three properties: id, name, and description. Furthermore, we have a simplistic Stored Procedure called INSERT_COFFEE that accepts three input parameters: id, name, and description. We also use a fully supported database. In that case, the following configuration for a stored procedure outbound adapter suffices:

<int-jdbc:stored-proc-outbound-channel-adapter data-source="dataSource"
    channel="insertCoffeeProcedureRequestChannel"
    stored-procedure-name="INSERT_COFFEE"/>

For more sophisticated options, consider passing in one or more ProcedureParameter values.

If you do provide ProcedureParameter values explicitly, by default, an ExpressionEvaluatingSqlParameterSourceFactory is used for parameter processing, to enable the full power of SpEL expressions.

If you need even more control over how parameters are retrieved, consider passing in a custom implementation of SqlParameterSourceFactory by using the sql-parameter-source-factory attribute.

The following listing calls out the attributes that matter for a stored procedure inbound channel adapter:

<int-jdbc:stored-proc-inbound-channel-adapter
                                   channel=""                                    
                                   stored-procedure-name=""
                                   data-source=""
                                   auto-startup="true"
                                   id=""
                                   ignore-column-meta-data="false"
                                   is-function="false"
                                   skip-undeclared-results=""                    
                                   return-value-required="false"                 
    <int:poller/>
    <int-jdbc:sql-parameter-definition name="" direction="IN"
                                               type="STRING"
                                               scale=""/>
    <int-jdbc:parameter name="" type="" value=""/>
    <int-jdbc:parameter name="" expression=""/>
    <int-jdbc:returning-resultset name="" row-mapper="" />
</int-jdbc:stored-proc-inbound-channel-adapter>

The following listing calls out the attributes that matter for a stored procedure outbound channel adapter:

<int-jdbc:stored-proc-outbound-channel-adapter channel=""                        
                                               stored-procedure-name=""
                                               data-source=""
                                               auto-startup="true"
                                               id=""
                                               ignore-column-meta-data="false"
                                               order=""                          
                                               sql-parameter-source-factory=""
                                               use-payload-as-parameter-source="">
    <int:poller fixed-rate=""/>
    <int-jdbc:sql-parameter-definition name=""/>
    <int-jdbc:parameter name=""/>

</int-jdbc:stored-proc-outbound-channel-adapter>

The following listing calls out the attributes that matter for a stored procedure outbound channel adapter:

<int-jdbc:stored-proc-outbound-gateway request-channel=""                        
                                       stored-procedure-name=""
                                       data-source=""
                                   auto-startup="true"
                                   id=""
                                   ignore-column-meta-data="false"
                                   is-function="false"
                                   order=""
                                   reply-channel=""                              
                                   reply-timeout=""                              
                                   return-value-required="false"                 
                                   skip-undeclared-results=""                    
                                   sql-parameter-source-factory=""
                                   use-payload-as-parameter-source="">
<int-jdbc:sql-parameter-definition name="" direction="IN"
                                   type=""
                                   scale="10"/>
<int-jdbc:sql-parameter-definition name=""/>
<int-jdbc:parameter name="" type="" value=""/>
<int-jdbc:parameter name="" expression=""/>
<int-jdbc:returning-resultset name="" row-mapper="" />

This section contains two examples that call Apache Derby stored procedures. The first procedure calls a stored procedure that returns a ResultSet. By using a RowMapper, the data is converted into a domain object, which then becomes the Spring Integration message payload.

In the second sample, we call a stored procedure that uses output parameters to return data instead.

	Note
	Have a look at the Spring Integration Samples project. The project contains the Apache Derby example referenced here, as well as instructions on how to run it. The Spring Integration Samples project also provides an example of using Oracle stored procedures.

In the first example, we call a stored procedure named FIND_ALL_COFFEE_BEVERAGES that does not define any input parameters but that returns a ResultSet.

In Apache Derby, stored procedures are implemented in Java. The following listing shows the method signature:

public static void findAllCoffeeBeverages(ResultSet[] coffeeBeverages)
            throws SQLException {
    ...
}

The following listing shows the corresponding SQL:

CREATE PROCEDURE FIND_ALL_COFFEE_BEVERAGES() \
PARAMETER STYLE JAVA LANGUAGE JAVA MODIFIES SQL DATA DYNAMIC RESULT SETS 1 \
EXTERNAL NAME 'o.s.i.jdbc.storedproc.derby.DerbyStoredProcedures.findAllCoffeeBeverages';

In Spring Integration, you can now call this stored procedure by using, for example, a stored-proc-outbound-gateway, as the following example shows:

<int-jdbc:stored-proc-outbound-gateway id="outbound-gateway-storedproc-find-all"
                                       data-source="dataSource"
                                       request-channel="findAllProcedureRequestChannel"
                                       expect-single-result="true"
                                       stored-procedure-name="FIND_ALL_COFFEE_BEVERAGES">
<int-jdbc:returning-resultset name="coffeeBeverages"
    row-mapper="org.springframework.integration.support.CoffeBeverageMapper"/>
</int-jdbc:stored-proc-outbound-gateway>

In the second example, we call a stored procedure named FIND_COFFEE that has one input parameter. Instead of returning a ResultSet, it uses an output parameter. The following example shows the method signature:

public static void findCoffee(int coffeeId, String[] coffeeDescription)
            throws SQLException {
    ...
}

The following listing shows the corresponding SQL:

CREATE PROCEDURE FIND_COFFEE(IN ID INTEGER, OUT COFFEE_DESCRIPTION VARCHAR(200)) \
PARAMETER STYLE JAVA LANGUAGE JAVA EXTERNAL NAME \
'org.springframework.integration.jdbc.storedproc.derby.DerbyStoredProcedures.findCoffee';

In Spring Integration, you can now call this Stored Procedure by using, for example, a stored-proc-outbound-gateway, as the following example shows:

<int-jdbc:stored-proc-outbound-gateway id="outbound-gateway-storedproc-find-coffee"
                                       data-source="dataSource"
                                       request-channel="findCoffeeProcedureRequestChannel"
                                       skip-undeclared-results="true"
                                       stored-procedure-name="FIND_COFFEE"
                                       expect-single-result="true">
    <int-jdbc:parameter name="ID" expression="payload" />
</int-jdbc:stored-proc-outbound-gateway>