Inbound Polling

package org.springframework.integration.jdbc.inbound;

public class JdbcPollingChannelAdapter extends AbstractMessageSource<Object> {
    public JdbcPollingChannelAdapter(DataSource dataSource, String selectQuery);
    public JdbcPollingChannelAdapter(JdbcOperations jdbcOperations, String selectQuery);

    public void setRowMapper(RowMapper<?> rowMapper);
    public void setSelectQuery(String selectQuery);
    public void setUpdateSql(String updateSql);
    public void setUpdatePerRow(boolean updatePerRow);
    public void setUpdateSqlParameterSourceFactory(SqlParameterSourceFactory factory);
    public void setSelectSqlParameterSource(SqlParameterSource sqlQueryParameterSource);
    public void setMaxRows(int maxRows);
    public List<?> doPoll(@Nullable SqlParameterSource sqlQueryParameterSource);
    public String getComponentType();
}

package org.springframework.integration.jdbc.inbound;

public class StoredProcPollingChannelAdapter extends AbstractMessageSource<Object> {
    public StoredProcPollingChannelAdapter(StoredProcExecutor storedProcExecutor);

    public void setExpectSingleResult(boolean expectSingleResult);
    public String getComponentType();
}

import org.springframework.integration.jdbc.inbound.JdbcPollingChannelAdapter;
import javax.sql.DataSource;

// Basic polling without updates
JdbcPollingChannelAdapter simpleAdapter = new JdbcPollingChannelAdapter(
    dataSource,
    "SELECT id, order_number, amount FROM orders WHERE status = 'PENDING'"
);
simpleAdapter.setMaxRows(100);

// Poll messages
List<?> messages = simpleAdapter.receive();

// Polling with per-row update marking
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
    dataSource,
    "SELECT id, order_number, amount FROM orders WHERE processed = false"
);
adapter.setUpdateSql("UPDATE orders SET processed = true WHERE id = :id");
adapter.setUpdatePerRow(true);
adapter.setMaxRows(50);

// Poll messages - each row updated individually
List<?> messages = adapter.receive();

import org.springframework.jdbc.core.BeanPropertyRowMapper;

// Custom row mapping to domain objects
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
    dataSource,
    "SELECT * FROM orders WHERE processed = false"
);
adapter.setRowMapper(new BeanPropertyRowMapper<>(Order.class));
adapter.setUpdateSql("UPDATE orders SET processed = true WHERE id = :id");
adapter.setUpdatePerRow(true);

// Poll returns List<Order>
List<?> orders = adapter.receive();

// Polling with batch update
JdbcPollingChannelAdapter batchAdapter = new JdbcPollingChannelAdapter(
    dataSource,
    "SELECT * FROM events WHERE status = 'NEW' ORDER BY created_at LIMIT 100"
);
batchAdapter.setUpdateSql("UPDATE events SET status = 'PROCESSED' WHERE id IN (:id)");
batchAdapter.setUpdatePerRow(false); // Batch update all polled records

import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;

// Using SqlParameterSource for SELECT query
SqlParameterSource params = new MapSqlParameterSource()
    .addValue("region", "US")
    .addValue("minAmount", 1000);

JdbcPollingChannelAdapter paramAdapter = new JdbcPollingChannelAdapter(
    dataSource,
    "SELECT * FROM orders WHERE region = :region AND amount > :minAmount"
);
paramAdapter.setSelectSqlParameterSource(params);

import org.springframework.integration.jdbc.StoredProcExecutor;
import org.springframework.integration.jdbc.inbound.StoredProcPollingChannelAdapter;
import org.springframework.integration.jdbc.storedproc.ProcedureParameter;

// Configure stored procedure executor
StoredProcExecutor executor = new StoredProcExecutor(dataSource);
executor.setStoredProcedureName("GET_PENDING_ORDERS");
executor.setIgnoreColumnMetaData(false);

// Define procedure parameters
List<ProcedureParameter> params = List.of(
    new ProcedureParameter("region", "US", null),
    new ProcedureParameter("limit", 100, null)
);
executor.setProcedureParameters(params);

// Create polling adapter
StoredProcPollingChannelAdapter adapter = new StoredProcPollingChannelAdapter(executor);

// Poll returns Map with result sets, OUT params, and return value
Message<?> message = adapter.receive();
// Payload is Map<String, Object> with keys: result set names, OUT param names, etc.

// Polling stored function
StoredProcExecutor functionExecutor = new StoredProcExecutor(dataSource);
functionExecutor.setStoredProcedureName("COUNT_PENDING_ORDERS");
functionExecutor.setIsFunction(true);
functionExecutor.setReturnValueRequired(true);

StoredProcPollingChannelAdapter funcAdapter = new StoredProcPollingChannelAdapter(functionExecutor);
funcAdapter.setExpectSingleResult(true);

Message<?> countMessage = funcAdapter.receive();
Integer pendingCount = (Integer) countMessage.getPayload();

import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.Pollers;
import static org.springframework.integration.jdbc.dsl.Jdbc.*;

@Bean
public IntegrationFlow pollingFlow(DataSource dataSource) {
    return IntegrationFlow
        .from(inboundAdapter(dataSource, "SELECT * FROM orders WHERE processed = false")
                .updateSql("UPDATE orders SET processed = true WHERE id = :id")
                .updatePerRow(true)
                .maxRows(100),
            e -> e.poller(Pollers.fixedDelay(5000))) // Poll every 5 seconds
        .handle(message -> {
            // Process message
            System.out.println("Received: " + message.getPayload());
        })
        .get();
}

<int-jdbc:inbound-channel-adapter
    channel="inputChannel"
    data-source="dataSource"
    query="SELECT * FROM orders WHERE processed = false"
    update="UPDATE orders SET processed = true WHERE id = :id"
    update-per-row="true"
    max-rows="100">
    <int:poller fixed-delay="5000"/>
</int-jdbc:inbound-channel-adapter>

import org.springframework.integration.jdbc.JdbcPollingChannelAdapter;
import org.springframework.integration.jdbc.StoredProcPollingChannelAdapter;

import org.springframework.integration.jdbc.inbound.JdbcPollingChannelAdapter;
import org.springframework.integration.jdbc.inbound.StoredProcPollingChannelAdapter;