<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:task="http://www.springframework.org/schema/task"
	xmlns:int="http://www.springframework.org/schema/integration"
	xsi:schemaLocation="http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
		http://www.springframework.org/schema/task https://www.springframework.org/schema/task/spring-task.xsd
		http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">

	<int:transaction-synchronization-factory id="syncFactory">
		<int:before-commit expression="@store.removeFromIdCache(headers.id.toString())"/>
		<int:after-commit channel="afterTxChannel"/>
	</int:transaction-synchronization-factory>

	<int:channel id="afterTxChannel">
		<int:queue/>
	</int:channel>

	<task:executor id="pool" pool-size="10" queue-capacity="10" rejection-policy="CALLER_RUNS" />

	<bean id="store" class="org.springframework.integration.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" />

	<bean id="testService" class="org.springframework.integration.jdbc.store.channel.TestService"/>

	<int:service-activator input-channel="outputChannel" ref="testService" method="process"/>

	<int:channel id="errorChannel">
		<int:queue/>
		<int:interceptors>
			<int:wire-tap channel="loggit"/>
		</int:interceptors>
	</int:channel>

	<int:logging-channel-adapter id="loggit" log-full-message="true" level="ERROR"/>

	<task:executor id="threadPoolTaskExecutor" pool-size="100"/>

	<bean id="messageStore" class="org.springframework.integration.jdbc.store.JdbcChannelMessageStore">
		<property name="dataSource" ref="dataSource"/>
		<property name="region" value="CONCURRENT_POLL"/>
		<property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
	</bean>

	<int:channel id="first">
		<int:queue message-store="messageStore"/>
	</int:channel>

	<int:channel id="second">
		<int:queue message-store="messageStore"/>
	</int:channel>

	<int:bridge input-channel="first" output-channel="second">
		<int:poller fixed-rate="200" task-executor="threadPoolTaskExecutor" max-messages-per-poll="3">
			<int:transactional transaction-manager="transactionManager"/>
		</int:poller>
	</int:bridge>

	<bean id="successfulLatch" class="java.util.concurrent.CountDownLatch">
		<constructor-arg value="10"/>
	</bean>

	<int:outbound-channel-adapter channel="second" expression="@successfulLatch.countDown()">
		<int:poller fixed-delay="100">
			<int:transactional transaction-manager="transactionManager"/>
		</int:poller>
	</int:outbound-channel-adapter>

	<bean id="errorAtomicInteger" class="java.util.concurrent.atomic.AtomicInteger"/>

	<int:service-activator input-channel="errorChannel" output-channel="nullChannel"
						   expression="@errorAtomicInteger.incrementAndGet()">
		<int:poller fixed-delay="1000"/>
	</int:service-activator>

	<bean id="priorityMessageStore" parent="messageStore">
		<property name="priorityEnabled" value="true"/>
	</bean>

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

</beans>

