INT-1518: Provide crutch for broken RDBMS (DB2, Derby etc.)

- Externalize storeLock and LockInterceptor utility
- Add tests showing usage of tx interceptor

.gitignore

@@ -1,4 +1,5 @@
 lib
+logs
 target
 .springBeans
 .settings

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java

@@ -36,7 +36,7 @@ import org.springframework.integration.Message;
  */
 public class MessageGroupQueue extends AbstractQueue<Message<?>> implements BlockingQueue<Message<?>> {
 
-	private static final int DEFAULT_CAPACITY = Integer.MAX_VALUE;
+	private static final int DEFAULT_CAPACITY = -1;
 
 	private final MessageGroupStore messageGroupStore;
 
@@ -45,13 +45,13 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 	private final int capacity;
 
 	// This one could be a global semaphore
-	private Object storeLock = new Object();
+	private volatile Object storeLock = new Object();
 
 	// This one only needs to be local
-	private Object writeLock = new Object();
+	private final Object writeLock = new Object();
 
 	// This one only needs to be local
-	private Object readLock = new Object();
+	private final Object readLock = new Object();
 
 	public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId) {
 		this(messageGroupStore, groupId, DEFAULT_CAPACITY);
@@ -62,6 +62,13 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		this.groupId = groupId;
 		this.capacity = capacity;
 	}
+	
+	/**
+	 * @param storeLock the storeLock to set
+	 */
+	public void setStoreLock(Object storeLock) {
+		this.storeLock = storeLock;
+	}
 
 	public Iterator<Message<?>> iterator() {
 		return getUnmarked().iterator();
@@ -73,7 +80,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 
 	public boolean offer(Message<?> e) {
 		synchronized (storeLock) {
-			if (messageGroupStore.getMessageGroup(groupId).size() >= capacity) {
+			if (capacity>0 && messageGroupStore.getMessageGroup(groupId).size() >= capacity) {
 				return false;
 			}
 			messageGroupStore.addMessageToGroup(groupId, e);
@@ -174,7 +181,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 	}
 
 	public int remainingCapacity() {
-		return capacity - messageGroupStore.getMessageGroup(groupId).size();
+		return (capacity>0 ? capacity : Integer.MAX_VALUE) - messageGroupStore.getMessageGroup(groupId).size();
 	}
 
 	public Message<?> take() throws InterruptedException {

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java

@@ -26,7 +26,6 @@ import javax.sql.DataSource;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.commons.serializer.Deserializer;
 import org.springframework.commons.serializer.DeserializingConverter;
 import org.springframework.commons.serializer.Serializer;
@@ -298,7 +297,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 	}
 
 	public MessageGroup getMessageGroup(Object groupId) {
-		String key = getKey(groupId);
+		String key = getKey(groupId);		
+		// TODO: collapse 3 queries into 1
 		List<Message<?>> marked = jdbcTemplate.query(getQuery(LIST_MARKED_MESSAGES_BY_GROUP_KEY), new Object[] {
 				key, region }, mapper);
 		List<Message<?>> unmarked = jdbcTemplate.query(getQuery(LIST_UNMARKED_MESSAGES_BY_GROUP_KEY),

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml

@@ -0,0 +1,72 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:beans="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
+	xmlns:tx="http://www.springframework.org/schema/tx"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd">
+
+	<jdbc:embedded-database id="dataSource" type="DERBY" />
+
+	<jdbc:initialize-database data-source="dataSource"
+		ignore-failures="DROPS">
+		<jdbc:script location="${int.drop.script}" />
+		<jdbc:script location="${int.schema.script}" />
+	</jdbc:initialize-database>
+
+	<int-jdbc:message-store id="messageStore"
+		data-source="dataSource" />
+
+	<channel id="input" xmlns="http://www.springframework.org/schema/integration">
+		<queue ref="storeQueue" />
+	</channel>
+
+	<bean id="storeQueue" class="org.springframework.integration.store.MessageGroupQueue">
+		<constructor-arg ref="messageStore" />
+		<constructor-arg value="input-queue" />
+		<property name="storeLock" ref="lock" />
+	</bean>
+
+	<int:channel id="output" />
+
+	<int:logging-channel-adapter channel="output" />
+
+	<service-activator id="service-activator"
+		input-channel="input" output-channel="output"
+		xmlns="http://www.springframework.org/schema/integration">
+		<beans:bean
+			class="org.springframework.integration.jdbc.JdbcMessageStoreChannelIntegrationTests$Service" />
+		<poller fixed-rate="200">
+			<advice-chain>
+				<ref bean="txAdvice" />
+				<ref bean="lock" />
+			</advice-chain>
+		</poller>
+	</service-activator>
+
+	<bean id="lock" class="org.springframework.integration.jdbc.LockInterceptor" />
+	<tx:advice id="txAdvice">
+		<tx:attributes>
+			<tx:method name="*" />
+		</tx:attributes>
+	</tx:advice>
+
+	<bean id="placeholderProperties"
+		class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
+		<property name="location"
+			value="classpath:int-${ENVIRONMENT:derby}.properties" />
+		<property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE" />
+		<property name="ignoreUnresolvablePlaceholders" value="true" />
+		<property name="order" value="1" />
+	</bean>
+
+	<bean id="transactionManager"
+		class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
+		<property name="dataSource" ref="dataSource" />
+	</bean>
+
+</beans>

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java

@@ -0,0 +1,214 @@
+/*
+ * Copyright 2002-2010 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.DefaultTransactionDefinition;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.util.StopWatch;
+
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class JdbcMessageStoreChannelIntegrationTests {
+
+	@Autowired
+	private QueueChannel input;
+
+	@Autowired
+	@Qualifier("lock")
+	private Object storeLock;
+
+	@Autowired
+	private JdbcMessageStore messageStore;
+
+	@Autowired
+	private PlatformTransactionManager transactionManager;
+
+	@Before
+	public void clear() {
+		for (MessageGroup group : messageStore) {
+			messageStore.removeMessageGroup(group.getGroupId());
+		}
+	}
+
+	@Test
+	public void testSendAndActivate() throws Exception {
+		Service.reset(1);
+		input.send(new GenericMessage<String>("foo"));
+		Service.await(1000);
+		assertEquals(1, Service.messages.size());
+	}
+
+	@Test
+	// @Repeat(50)
+	public void testSendAndActivateWithRollback() throws Exception {
+		Service.reset(1);
+		Service.fail = true;
+		input.send(new GenericMessage<String>("foo"));
+		Service.await(1000);
+		assertEquals(1, Service.messages.size());
+		// After a rollback in the poller the message is still waiting to be delivered
+		assertEquals(1, input.getQueueSize());
+		assertNotNull(input.receive(100L));
+	}
+
+	@Test
+	public void testTransactionalSendAndReceive() throws Exception {
+
+		Service.reset(1);
+
+		boolean result = new TransactionTemplate(transactionManager).execute(new TransactionCallback<Boolean>() {
+
+			public Boolean doInTransaction(TransactionStatus status) {
+
+				synchronized (storeLock) {
+
+					boolean result = input.send(new GenericMessage<String>("foo"), 500L);
+					// This will time out because the transaction has not committed yet
+					try {
+						Service.await(1000);
+						fail("Expected timeout");
+					} catch (Exception e) {
+						// expected
+					}
+
+					return result;
+
+				}
+
+			}
+		});
+
+		assertTrue("Could not send message", result);
+
+		// So no activation
+		assertEquals(0, Service.messages.size());
+
+		StopWatch stopWatch = new StopWatch();
+		try {
+			stopWatch.start();
+			// It might be null or not, but we don't want it to block
+			input.receive(100L);
+		} finally {
+			stopWatch.stop();
+		}
+
+		// If the poll blocks in the RDBMS there is no way for the queue to respect the timeout
+		assertTrue("Timed out waiting for receive", stopWatch.getTotalTimeMillis() < 10000);
+
+	}
+
+	@Test
+	public void testSameTransactionSendAndReceive() throws Exception {
+
+		Service.reset(1);
+		final StopWatch stopWatch = new StopWatch();
+		DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
+
+		// With a timeout on the transaction the test fails (after a long time) on the assertion in the transactional
+		// receive.
+		transactionDefinition.setTimeout(200);
+
+		boolean result = new TransactionTemplate(transactionManager, transactionDefinition)
+				.execute(new TransactionCallback<Boolean>() {
+
+					public Boolean doInTransaction(TransactionStatus status) {
+
+						synchronized (storeLock) {
+
+							boolean result = input.send(new GenericMessage<String>("foo"), 500L);
+							// This will time out because the transaction has not committed yet
+							try {
+								Service.await(1000);
+								fail("Expected timeout");
+							} catch (Exception e) {
+								// expected
+							}
+
+							try {
+								stopWatch.start();
+								assertNotNull(input.receive(100L));
+							} finally {
+								stopWatch.stop();
+							}
+
+							return result;
+
+						}
+
+					}
+				});
+
+		assertTrue("Could not send message", result);
+
+		// So no activation
+		assertEquals(0, Service.messages.size());
+
+		// If the poll blocks in the RDBMS there is no way for the queue to respect the timeout
+		assertTrue("Timed out waiting for receive", stopWatch.getTotalTimeMillis() < 1000);
+
+	}
+
+	public static class Service {
+		private static boolean fail = false;
+
+		private static List<String> messages = new CopyOnWriteArrayList<String>();
+
+		private static CountDownLatch latch = new CountDownLatch(0);
+
+		public static void reset(int count) {
+			fail = false;
+			messages.clear();
+			latch = new CountDownLatch(count);
+		}
+
+		public static void await(long timeout) throws InterruptedException {
+			if (!latch.await(timeout, TimeUnit.MILLISECONDS)) {
+				throw new IllegalStateException("Timed out waiting for message");
+			}
+		}
+
+		public String echo(String input) {
+			messages.add(input);
+			latch.countDown();
+			if (fail) {
+				throw new RuntimeException("Planned failure");
+			}
+			return input;
+		}
+	}
+
+}

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml

@@ -0,0 +1,63 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
+	xmlns:tx="http://www.springframework.org/schema/tx"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd">
+
+	<jdbc:embedded-database id="dataSource" type="DERBY" />
+
+	<jdbc:initialize-database data-source="dataSource"
+		ignore-failures="DROPS">
+		<jdbc:script location="${int.drop.script}" />
+		<jdbc:script location="${int.schema.script}" />
+	</jdbc:initialize-database>
+
+	<int-jdbc:message-store id="messageStore"
+		data-source="dataSource" />
+
+	<channel id="relay" xmlns="http://www.springframework.org/schema/integration">
+		<queue message-store="messageStore" />
+	</channel>
+
+	<channel id="durable" xmlns="http://www.springframework.org/schema/integration">
+		<queue message-store="messageStore" />
+	</channel>
+
+	<service-activator id="service-relay" input-channel="relay"
+		output-channel="durable" xmlns="http://www.springframework.org/schema/integration">
+		<beans:bean
+			class="org.springframework.integration.jdbc.JdbcMessageStoreChannelOnePollerIntegrationTests$Service" />
+		<poller fixed-rate="200">
+			<advice-chain>
+				<ref bean="lock" />
+				<ref bean="txAdvice" />
+			</advice-chain>
+		</poller>
+	</service-activator>
+
+	<bean id="lock" class="org.springframework.integration.jdbc.LockInterceptor" />
+	<tx:advice id="txAdvice"/>
+
+	<bean id="placeholderProperties"
+		class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
+		<property name="location"
+			value="classpath:int-${ENVIRONMENT:derby}.properties" />
+		<property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE" />
+		<property name="ignoreUnresolvablePlaceholders" value="true" />
+		<property name="order" value="1" />
+	</bean>
+
+	<bean id="transactionManager"
+		class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
+		<property name="dataSource" ref="dataSource" />
+	</bean>
+
+</beans>

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java

@@ -0,0 +1,176 @@
+/*
+ * Copyright 2002-2010 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.util.StopWatch;
+
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class JdbcMessageStoreChannelOnePollerIntegrationTests {
+
+	@Autowired
+	private QueueChannel relay;
+
+	@Autowired
+	private QueueChannel durable;
+
+	@Autowired
+	@Qualifier("lock")
+	private Object storeLock;
+
+	@Autowired
+	private JdbcMessageStore messageStore;
+
+	@Autowired
+	private PlatformTransactionManager transactionManager;
+
+	@Before
+	public void clear() {
+		for (MessageGroup group : messageStore) {
+			messageStore.removeMessageGroup(group.getGroupId());
+		}
+	}
+
+	@Test
+	// @Repeat(50)
+	public void testSameTransactionDifferentChannelSendAndReceive() throws Exception {
+
+		Service.reset(1);
+		assertNull(durable.receive(100L));
+		assertNull(relay.receive(100L));
+		final StopWatch stopWatch = new StopWatch();
+
+		boolean result = new TransactionTemplate(transactionManager).execute(new TransactionCallback<Boolean>() {
+
+			public Boolean doInTransaction(TransactionStatus status) {
+
+				synchronized (storeLock) {
+
+					boolean result = relay.send(new GenericMessage<String>("foo"), 500L);
+					// This will time out because the transaction has not committed yet
+					try {
+						Service.await(1000);
+						fail("Expected timeout");
+					} catch (Exception e) {
+						// expected
+					}
+
+					try {
+						stopWatch.start();
+						// It hasn't arrive yet because we are still in the sending transaction
+						assertNull(durable.receive(100L));
+					} finally {
+						stopWatch.stop();
+					}
+
+					return result;
+
+				}
+
+			}
+		});
+
+		assertTrue("Could not send message", result);
+		// If the poll blocks in the RDBMS there is no way for the queue to respect the timeout
+		assertTrue("Timed out waiting for receive", stopWatch.getTotalTimeMillis() < 10000);
+
+		Service.await(1000);
+		// Eventual activation
+		assertEquals(1, Service.messages.size());
+
+		/*
+		 * Without the storeLock:
+		 * 
+		 * If we do this in a transaction it deadlocks occasionally. Without a transaction and it's pretty much every
+		 * time.
+		 * 
+		 * With the storeLock: It doesn't deadlock as long as the lock is injected into the poller as well.
+		 */
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
+
+			public Void doInTransaction(TransactionStatus status) {
+				synchronized (storeLock) {
+
+					try {
+						stopWatch.start();
+						durable.receive(100L);
+						return null;
+					} finally {
+						stopWatch.stop();
+					}
+
+				}
+			}
+
+		});
+
+		// If the poll blocks in the RDBMS there is no way for the queue to respect the timeout
+		assertTrue("Timed out waiting for receive", stopWatch.getTotalTimeMillis() < 10000);
+
+	}
+
+	public static class Service {
+		private static boolean fail = false;
+
+		private static List<String> messages = new CopyOnWriteArrayList<String>();
+
+		private static CountDownLatch latch = new CountDownLatch(0);
+
+		public static void reset(int count) {
+			fail = false;
+			messages.clear();
+			latch = new CountDownLatch(count);
+		}
+
+		public static void await(long timeout) throws InterruptedException {
+			if (!latch.await(timeout, TimeUnit.MILLISECONDS)) {
+				throw new IllegalStateException("Timed out waiting for message");
+			}
+		}
+
+		public String echo(String input) {
+			messages.add(input);
+			latch.countDown();
+			if (fail) {
+				throw new RuntimeException("Planned failure");
+			}
+			return input;
+		}
+	}
+
+}

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests-context.xml

@@ -32,8 +32,7 @@
 
 	<service-activator id="service-activator" input-channel="input" output-channel="output" xmlns="http://www.springframework.org/schema/integration">
 		<beans:bean class="org.springframework.integration.jdbc.JdbcMessageStoreChannelTests$Service" />
-		<poller>
-			<interval-trigger interval="200"/>
+		<poller fixed-rate="200">
 			<transactional />
 		</poller>
 	</service-activator>

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/LockInterceptor.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2002-2010 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.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class LockInterceptor implements MethodInterceptor {
+
+	public synchronized Object invoke(MethodInvocation invocation) throws Throwable {
+		return invocation.proceed();
+	}
+
+}