Added TransactionInterceptorTests.

org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TestBean.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2002-2008 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.endpoint.interceptor;
+
+/**
+ * @author Mark Fisher
+ */
+public class TestBean {
+
+	public String good(String s) {
+		return s;
+	}
+
+	public String bad(String s) {
+		throw new RuntimeException("intentional test failure");
+	}
+
+}

org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TestTransactionManager.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2002-2008 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.endpoint.interceptor;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionException;
+import org.springframework.transaction.support.AbstractPlatformTransactionManager;
+import org.springframework.transaction.support.DefaultTransactionDefinition;
+import org.springframework.transaction.support.DefaultTransactionStatus;
+
+/**
+ * @author Mark Fisher
+ */
+@SuppressWarnings("serial")
+public class TestTransactionManager extends AbstractPlatformTransactionManager {
+
+	private final AtomicInteger commitCounter = new AtomicInteger();
+
+	private final AtomicInteger rollbackCounter = new AtomicInteger();
+
+	private final CountDownLatch latch = new CountDownLatch(1);
+
+
+	public int getCommitCount() {
+		return this.commitCounter.get();
+	}
+
+	public int getRollbackCount() {
+		return this.rollbackCounter.get();
+	}
+
+	public void waitForCompletion(long timeout) throws InterruptedException {
+		this.latch.await(timeout, TimeUnit.MILLISECONDS);
+	}
+
+	@Override
+	protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
+	}
+
+	@Override
+	protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
+		this.commitCounter.incrementAndGet();
+	}
+
+	@Override
+	protected Object doGetTransaction() throws TransactionException {
+		return new DefaultTransactionDefinition();
+	}
+
+	@Override
+	protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
+		this.rollbackCounter.incrementAndGet();
+	}
+
+	@Override
+	protected void doCleanupAfterCompletion(Object o) {
+		this.latch.countDown();
+	}
+
+}

org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2002-2008 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.endpoint.interceptor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import org.junit.Test;
+
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.StringMessage;
+
+/**
+ * @author Mark Fisher
+ */
+public class TransactionInterceptorTests {
+
+	@Test
+	public void testTransactionInterceptorWithCommit() throws InterruptedException {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"transactionInterceptorTests.xml", this.getClass());
+		TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
+		MessageChannel input = (MessageChannel) context.getBean("goodInput");
+		MessageChannel output = (MessageChannel) context.getBean("output");
+		assertEquals(0, txManager.getCommitCount());
+		assertEquals(0, txManager.getRollbackCount());
+		input.send(new StringMessage("test"));
+		txManager.waitForCompletion(1000);
+		Message<?> message = output.receive(0);
+		assertNotNull(message);		
+		assertEquals(1, txManager.getCommitCount());
+		assertEquals(0, txManager.getRollbackCount());		
+	}
+
+	@Test
+	public void testTransactionInterceptorWithRollback() throws InterruptedException {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"transactionInterceptorTests.xml", this.getClass());
+		TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
+		MessageChannel input = (MessageChannel) context.getBean("badInput");
+		MessageChannel output = (MessageChannel) context.getBean("output");
+		assertEquals(0, txManager.getCommitCount());
+		assertEquals(0, txManager.getRollbackCount());
+		input.send(new StringMessage("test"));
+		txManager.waitForCompletion(1000);
+		Message<?> message = output.receive(0);
+		assertNull(message);
+		assertEquals(0, txManager.getCommitCount());
+		assertEquals(1, txManager.getRollbackCount());		
+	}
+
+}

org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorTests.xml

@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<message-bus/>
+
+	<channel id="badInput"/>
+	<channel id="goodInput"/>
+	<channel id="output"/>
+
+	<handler-endpoint input-channel="badInput"
+	                  handler="testBean"
+	                  method="bad"
+                      output-channel="output">
+        <schedule period="100"/>
+    	<interceptors>
+			<transaction-interceptor transaction-manager="txManager"/>
+    	</interceptors>
+    </handler-endpoint>
+
+	<handler-endpoint input-channel="goodInput"
+	                  handler="testBean"
+	                  method="good"
+                      output-channel="output">
+        <schedule period="100"/>
+		<interceptors>
+			<transaction-interceptor transaction-manager="txManager" propagation="REQUIRED"/>
+		</interceptors>
+	</handler-endpoint>
+
+	<beans:bean id="testBean" class="org.springframework.integration.endpoint.interceptor.TestBean"/>
+
+	<beans:bean id="txManager" class="org.springframework.integration.endpoint.interceptor.TestTransactionManager"/>
+
+</beans:beans>