diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/AmqpConnectException.java b/spring-amqp-core/src/main/java/org/springframework/amqp/AmqpConnectException.java index 93ede5d2..06bc8a07 100644 --- a/spring-amqp-core/src/main/java/org/springframework/amqp/AmqpConnectException.java +++ b/spring-amqp-core/src/main/java/org/springframework/amqp/AmqpConnectException.java @@ -22,9 +22,9 @@ import java.net.ConnectException; * @author Dave Syer */ @SuppressWarnings("serial") -public class AmqpConnectException extends AmqpIOException { +public class AmqpConnectException extends AmqpException { - public AmqpConnectException(ConnectException cause) { + public AmqpConnectException(Exception cause) { super(cause); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java index a0b38093..d90621a7 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java @@ -102,7 +102,7 @@ public class ConnectionFactoryUtils { } }); if (synchedLocalTransactionAllowed) { - holder.declareTransactional(); + // holder.declareTransactional(); } return holder; } @@ -132,43 +132,27 @@ public class ConnectionFactoryUtils { if (resourceHolderToUse == null) { resourceHolderToUse = new RabbitResourceHolder(); } - Connection con = resourceFactory.getConnection(resourceHolderToUse); + Connection connection = resourceFactory.getConnection(resourceHolderToUse); Channel channel = null; try { - boolean isExistingCon = (con != null); + boolean isExistingCon = (connection != null); if (!isExistingCon) { - con = resourceFactory.createConnection(); - resourceHolderToUse.addConnection(con); + connection = resourceFactory.createConnection(); + resourceHolderToUse.addConnection(connection); } - channel = resourceFactory.createChannel(con); - resourceHolderToUse.addChannel(channel, con); + channel = resourceFactory.createChannel(connection); + resourceHolderToUse.addChannel(channel, connection); if (resourceHolderToUse != resourceHolder) { bindResourceToTransaction(resourceHolderToUse, connectionFactory, resourceFactory.isSynchedLocalTransactionAllowed()); } - if (isChannelTransactional(channel, connectionFactory)) { - // It is externally transacted and was just created so we want to start the transaction - channel.txSelect(); - } return resourceHolderToUse; } catch (IOException ex) { - if (channel != null) { - try { - channel.close(); - } catch (Throwable ex2) { - // ignore - } - } - if (con != null) { - try { - con.close(); - } catch (Throwable ex2) { - // ignore - } - } + RabbitUtils.closeChannel(channel); + RabbitUtils.closeConnection(connection); throw new AmqpIOException(ex); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java index 1229381f..b942551e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java @@ -17,7 +17,6 @@ import java.io.IOException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.amqp.AmqpException; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; @@ -103,7 +102,7 @@ public abstract class RabbitAccessor implements InitializingBean { return holder; } - protected AmqpException convertRabbitAccessException(Exception ex) { + protected RuntimeException convertRabbitAccessException(Exception ex) { return RabbitUtils.convertRabbitAccessException(ex); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java index 4458de4e..c71c1c46 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java @@ -36,6 +36,7 @@ import com.rabbitmq.client.AMQP; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Envelope; +import com.rabbitmq.client.ShutdownSignalException; /** * @author Mark Fisher @@ -101,11 +102,14 @@ public abstract class RabbitUtils { } } - public static AmqpException convertRabbitAccessException(Throwable ex) { + public static RuntimeException convertRabbitAccessException(Throwable ex) { Assert.notNull(ex, "Exception must not be null"); if (ex instanceof AmqpException) { return (AmqpException) ex; } + if (ex instanceof ShutdownSignalException) { + return new AmqpConnectException((ShutdownSignalException) ex); + } if (ex instanceof ConnectException) { return new AmqpConnectException((ConnectException) ex); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java index 2c819b9f..d0e9953a 100755 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java @@ -1,23 +1,19 @@ /* * 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. + * + * 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.amqp.rabbit.connection; import java.io.IOException; - import com.rabbitmq.client.Channel; public class SimpleConnection implements Connection { @@ -25,13 +21,18 @@ public class SimpleConnection implements Connection { private final com.rabbitmq.client.Connection delegate; public SimpleConnection(com.rabbitmq.client.Connection delegate) { - this.delegate = delegate; + this.delegate = delegate; } // TODO: expose the transactional flag public Channel createChannel(boolean transactional) { try { - return delegate.createChannel(); + Channel channel = delegate.createChannel(); + if (transactional) { + // Just created so we want to start the transaction + channel.txSelect(); + } + return channel; } catch (IOException e) { throw RabbitUtils.convertRabbitAccessException(e); } @@ -46,8 +47,7 @@ public class SimpleConnection implements Connection { } public boolean isOpen() { - return delegate!=null && delegate.isOpen(); + return delegate != null && delegate.isOpen(); } - } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 288b5f66..303de08f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -312,11 +312,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { if (isChannelLocallyTransacted(channel)) { resourceHolder.rollbackAll(); } - if (ex instanceof RuntimeException) { - throw (RuntimeException) ex; - } else { - throw convertRabbitAccessException(ex); - } + throw convertRabbitAccessException(ex); } finally { ConnectionFactoryUtils.releaseResources(resourceHolder); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java index d7a8ff6b..e4778db3 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java @@ -141,7 +141,7 @@ public class RabbitTransactionManager extends AbstractPlatformTransactionManager if (logger.isDebugEnabled()) { logger.debug("Created AMQP transaction on channel [" + resourceHolder.getChannel() + "]"); } - resourceHolder.declareTransactional(); + // resourceHolder.declareTransactional(); txObject.setResourceHolder(resourceHolder); txObject.getResourceHolder().setSynchronizedWithTransaction(true); int timeout = determineTimeout(definition); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 0f0ce7c0..403dfbed 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -24,14 +24,17 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import org.junit.Before; import org.junit.Rule; import org.junit.Test; +import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.RabbitUtils; import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionException; @@ -48,11 +51,36 @@ public class RabbitTemplateIntegrationTests { private static final String ROUTE = "test.queue"; - private RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory()); + private RabbitTemplate template; + + @Before + public void create() { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + connectionFactory.setPort(BrokerTestUtils.getPort()); + template = new RabbitTemplate(connectionFactory); + } @Rule public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(ROUTE); + @Test + public void testSendToNonExistentAndThenReceive() throws Exception { + // If transacted then the commit fails on send, so we get a nice synchronous exception + template.setChannelTransacted(true); + try { + template.convertAndSend("", "no.such.route", "message"); + // fail("Expected AmqpException"); + } catch (AmqpException e) { + // e.printStackTrace(); + } + // Now send the real message, and all should be well... + template.convertAndSend(ROUTE, "message"); + String result = (String) template.receiveAndConvert(ROUTE); + assertEquals("message", result); + result = (String) template.receiveAndConvert(ROUTE); + assertEquals(null, result); + } + @Test public void testSendAndReceive() throws Exception { template.convertAndSend(ROUTE, "message"); @@ -103,7 +131,7 @@ public class RabbitTemplateIntegrationTests { }); fail("Expected PlannedException"); } catch (Exception e) { - assertTrue(e instanceof PlannedException); + assertTrue(e.getCause() instanceof PlannedException); } String result = (String) template.receiveAndConvert(ROUTE); assertEquals("message", result);