From e0dcf08b7e30b415d4b0cb848c774a93b8165769 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Sun, 19 Sep 2010 09:49:44 +0100 Subject: [PATCH] INT-1518: Provide crutch for broken RDBMS (DB2, Derby etc.) - Externalize storeLock and LockInterceptor utility - Add tests showing usage of tx interceptor --- .gitignore | 1 + .../integration/store/MessageGroupQueue.java | 19 +- .../integration/jdbc/JdbcMessageStore.java | 4 +- ...geStoreChannelIntegrationTests-context.xml | 72 ++++++ ...bcMessageStoreChannelIntegrationTests.java | 214 ++++++++++++++++++ ...annelOnePollerIntegrationTests-context.xml | 63 ++++++ ...StoreChannelOnePollerIntegrationTests.java | 176 ++++++++++++++ .../JdbcMessageStoreChannelTests-context.xml | 3 +- .../integration/jdbc/LockInterceptor.java | 29 +++ 9 files changed, 571 insertions(+), 10 deletions(-) create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/LockInterceptor.java diff --git a/.gitignore b/.gitignore index 8eedbd10e5..a085e36f76 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ lib +logs target .springBeans .settings diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java index 30f08365e5..efd0d320fb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java +++ b/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> implements BlockingQueue> { - 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> 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> implements Bloc this.groupId = groupId; this.capacity = capacity; } + + /** + * @param storeLock the storeLock to set + */ + public void setStoreLock(Object storeLock) { + this.storeLock = storeLock; + } public Iterator> iterator() { return getUnmarked().iterator(); @@ -73,7 +80,7 @@ public class MessageGroupQueue extends AbstractQueue> 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> 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 { diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index 6ef9206afe..c57e5e679a 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/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> marked = jdbcTemplate.query(getQuery(LIST_MARKED_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper); List> unmarked = jdbcTemplate.query(getQuery(LIST_UNMARKED_MESSAGES_BY_GROUP_KEY), diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml new file mode 100644 index 0000000000..7465176f06 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml @@ -0,0 +1,72 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java new file mode 100644 index 0000000000..39cb40d981 --- /dev/null +++ b/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("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("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() { + + public Boolean doInTransaction(TransactionStatus status) { + + synchronized (storeLock) { + + boolean result = input.send(new GenericMessage("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() { + + public Boolean doInTransaction(TransactionStatus status) { + + synchronized (storeLock) { + + boolean result = input.send(new GenericMessage("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 messages = new CopyOnWriteArrayList(); + + 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; + } + } + +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml new file mode 100644 index 0000000000..af5746cc24 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml @@ -0,0 +1,63 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java new file mode 100644 index 0000000000..79c702b36c --- /dev/null +++ b/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() { + + public Boolean doInTransaction(TransactionStatus status) { + + synchronized (storeLock) { + + boolean result = relay.send(new GenericMessage("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() { + + 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 messages = new CopyOnWriteArrayList(); + + 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; + } + } + +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests-context.xml index 551e92cfb3..997a101146 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests-context.xml @@ -32,8 +32,7 @@ - - + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/LockInterceptor.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/LockInterceptor.java new file mode 100644 index 0000000000..5a5adfd18a --- /dev/null +++ b/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(); + } + +}