Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-136: move txSelect() to Connection impl

Browse Source
This commit is contained in:
Dave Syer
2011-03-28 18:16:59 +01:00
parent 7d70286dba
commit 7d57669c4c
8 changed files with 66 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-amqp-core/src/main/java/org/springframework/amqp/AmqpConnectException.java
View File

@@ -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);
}

34
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java
View File

@@ -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);
}
}

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java
View File

@@ -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);
}

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java
View File

@@ -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);
}

34
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java
View File

@@ -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();
}
}

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View File

@@ -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);
}

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/transaction/RabbitTransactionManager.java
View File

@@ -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);

32
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
View File

@@ -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);