* The property name will be the camel-case equivalent of the lower case hyphen separated attribute (e.g. the * "foo-bar" attribute would match the "fooBar" property). - * + * * @see Conventions#attributeNameToPropertyName(String) - * + * * @param builder the bean definition builder to be configured * @param element the XML element where the attribute should be defined * @param attributeName the name of the attribute whose value will be set on the property @@ -81,7 +81,7 @@ public abstract class NamespaceUtils { /** * Checks the attribute to see if it is defined in the given element. - * + * * @param element the XML element where the attribute should be defined * @param attributeName the name of the attribute whose value will be used as a constructor argument */ @@ -93,7 +93,7 @@ public abstract class NamespaceUtils { /** * Populates the bean definition constructor argument with the value of that attribute if it is defined in the given * element. - * + * * @param builder the bean definition builder to be configured * @param element the XML element where the attribute should be defined * @param attributeName the name of the attribute whose value will be used as a constructor argument @@ -111,7 +111,7 @@ public abstract class NamespaceUtils { /** * Populates the bean definition constructor argument with the boolean value of that attribute if it is defined in * the given element or else uses the default provided. - * + * * @param builder the bean definition builder to be configured * @param element the XML element where the attribute should be defined * @param attributeName the name of the attribute whose value will be used as a constructor argument @@ -130,7 +130,7 @@ public abstract class NamespaceUtils { /** * Populates the bean definition constructor argument with a reference to a bean with id equal to the attribute if * it is defined in the given element. - * + * * @param builder the bean definition builder to be configured * @param element the XML element where the attribute should be defined * @param attributeName the name of the attribute whose value will be used to set the reference @@ -148,7 +148,7 @@ public abstract class NamespaceUtils { /** * Populates the bean definition constructor argument with a reference to a bean with parent id equal to the * attribute if it is defined in the given element. - * + * * @param builder the bean definition builder to be configured * @param element the XML element where the attribute should be defined * @param attributeName the name of the attribute whose value will be used to set the reference @@ -168,13 +168,13 @@ public abstract class NamespaceUtils { /** * Populates the specified bean definition property with the reference to a bean. The bean reference is identified * by the value from the attribute whose name is provided if that attribute is defined in the given element. - * + * * @param builder the bean definition builder to be configured * @param element the XML element where the attribute should be defined * @param attributeName the name of the attribute whose value will be used as a bean reference to populate the * property * @param propertyName the name of the property to be populated - * @return + * @return true if the attribute is present and has text */ public static boolean setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName, String propertyName) { @@ -189,18 +189,18 @@ public abstract class NamespaceUtils { /** * Populates the bean definition property corresponding to the specified attributeName with the reference to a bean * identified by the value of that attribute if the attribute is defined in the given element. - * + * *
* The property name will be the camel-case equivalent of the lower case hyphen separated attribute (e.g. the * "foo-bar" attribute would match the "fooBar" property). - * + * * @see Conventions#attributeNameToPropertyName(String) - * + * * @param builder the bean definition builder to be configured * @param element the XML element where the attribute should be defined * @param attributeName the name of the attribute whose value will be used as a bean reference to populate the * property - * + * * @see Conventions#attributeNameToPropertyName(String) */ public static boolean setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element, diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 70cd916e..58d047dd 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -1,11 +1,11 @@ /* * Copyright 2002-2012 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. @@ -33,17 +33,17 @@ import com.rabbitmq.client.Channel; * A {@link ConnectionFactory} implementation that returns the same Connections from all {@link #createConnection()} * calls, and ignores calls to {@link com.rabbitmq.client.Connection#close()} and caches * {@link com.rabbitmq.client.Channel}. - * + * *
* By default, only one Channel will be cached, with further requested Channels being created and disposed on demand. * Consider raising the {@link #setChannelCacheSize(int) "channelCacheSize" value} in case of a high-concurrency * environment. - * + * *
* NOTE: This ConnectionFactory requires explicit closing of all Channels obtained form its shared Connection. * This is the usual recommendation for native Rabbit access code anyway. However, with this ConnectionFactory, its use * is mandatory in order to actually allow for Channel reuse. - * + * * @author Mark Pollack * @author Mark Fisher * @author Dave Syer @@ -77,9 +77,11 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { } /** - * Create a new CachingConnectionFactory given a host name. - * + * Create a new CachingConnectionFactory given a host name + * and port. + * * @param hostname the host name to connect to + * @param port the port number */ public CachingConnectionFactory(String hostname, int port) { super(new com.rabbitmq.client.ConnectionFactory()); @@ -91,9 +93,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { } /** - * Create a new CachingConnectionFactory given a host name. - * - * @param hostName the host name to connect to + * Create a new CachingConnectionFactory given a port on the hostname returned from + * InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception. + * + * @param port the port number */ public CachingConnectionFactory(int port) { this(null, port); @@ -101,7 +104,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { /** * Create a new CachingConnectionFactory given a host name. - * + * * @param hostname the host name to connect to */ public CachingConnectionFactory(String hostname) { @@ -110,7 +113,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { /** * Create a new CachingConnectionFactory for the given target ConnectionFactory. - * + * * @param rabbitConnectionFactory the target ConnectionFactory */ public CachingConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory) { @@ -142,6 +145,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { this.publisherConfirms = publisherConfirms; } + @Override public void setConnectionListeners(List listeners) { super.setConnectionListeners(listeners); // If the connection is already alive we assume that the new listeners want to be notified @@ -150,6 +154,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { } } + @Override public void addConnectionListener(ConnectionListener listener) { super.addConnectionListener(listener); // If the connection is already alive we assume that the new listener wants to be notified @@ -234,6 +239,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { * As this bean implements DisposableBean, a bean factory will automatically invoke this on destruction of its * cached singletons. */ + @Override public final void destroy() { synchronized (this.connectionMonitor) { if (connection != null) { @@ -358,7 +364,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { /** * GUARDED by channelList - * + * * @param proxy the channel to close */ private void logicalClose(ChannelProxy proxy) throws Exception { 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 6fa32bda..7fef25ac 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 @@ -1,11 +1,11 @@ /* * Copyright 2002-2012 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. @@ -61,14 +61,14 @@ import com.rabbitmq.client.GetResponse; *
* Helper class that simplifies synchronous RabbitMQ access (sending and receiving messages). *
- * + * *
* The default settings are for non-transactional messaging, which reduces the amount of data exchanged with the broker.
* To use a new transaction for every send or receive set the {@link #setChannelTransacted(boolean) channelTransacted}
* flag. To extend the transaction over multiple invocations (more efficient), you can use a Spring transaction to
* bracket the calls (with channelTransacted=true as well).
*
* The only mandatory property is the {@link #setConnectionFactory(ConnectionFactory) ConnectionFactory}. There are * strategies available for converting messages to and from Java objects ( @@ -77,7 +77,7 @@ import com.rabbitmq.client.GetResponse; * ). The defaults probably do something sensible for typical use cases, as long as the message content-type is set * appropriately. *
- * + * ** The "send" methods all have overloaded versions that allow you to explicitly target an exchange and a routing key, or * you can set default values to be used in all send operations. The plain "receive" methods allow you to explicitly @@ -85,7 +85,7 @@ import com.rabbitmq.client.GetResponse; * receives. The convenience methods for send and receive use the sender defaults if no exchange or routing key * is specified, but they always use a temporary queue for the receive leg, so the default queue is ignored. *
- * + * * @author Mark Pollack * @author Mark Fisher * @author Dave Syer @@ -147,7 +147,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, /** * Create a rabbit template with default strategies and settings. - * + * * @param connectionFactory the connection factory to use */ public RabbitTemplate(ConnectionFactory connectionFactory) { @@ -166,7 +166,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, /** * The name of the default exchange to use for send operations when none is specified. Defaults to""
* which is the default exchange in the broker (per the AMQP specification).
- *
+ *
* @param exchange the exchange name to use for send operations
*/
public void setExchange(String exchange) {
@@ -177,8 +177,8 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
* The value of a default routing key to use for send operations when none is specified. Default is empty which is
* not helpful when using the default (or any direct) exchange, but fine if the exchange is a headers exchange for
* instance.
- *
- * @param exchange the default routing key to use for send operations
+ *
+ * @param routingKey the default routing key to use for send operations
*/
public void setRoutingKey(String routingKey) {
this.routingKey = routingKey;
@@ -186,7 +186,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
/**
* The name of the default queue to receive messages from when none is specified explicitly.
- *
+ *
* @param queue the default queue name to use for receive
*/
public void setQueue(String queue) {
@@ -195,7 +195,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
/**
* The encoding to use when inter-converting between byte arrays and Strings in message properties.
- *
+ *
* @param encoding the encoding to set
*/
public void setEncoding(String encoding) {
@@ -218,9 +218,9 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
* sendAndReceive methods. The default value is defined as {@link #DEFAULT_REPLY_TIMEOUT}. A negative value
* indicates an indefinite timeout. Not used in the plain receive methods because there is no blocking receive
* operation defined in the protocol.
- *
+ *
* @param replyTimeout the reply timeout in milliseconds
- *
+ *
* @see #sendAndReceive(String, String, Message)
* @see #convertSendAndReceive(String, String, Object)
*/
@@ -234,7 +234,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
* * The default converter is a SimpleMessageConverter, which is able to handle byte arrays, Strings, and Serializable * Objects depending on the message content type header. - * + * * @see #convertAndSend * @see #receiveAndConvert * @see org.springframework.amqp.support.converter.SimpleMessageConverter @@ -484,7 +484,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, /** * Send a message and wait for a reply. - * + * * @param exchange the exchange name * @param routingKey the routing key * @param message the message to send @@ -610,7 +610,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, /** * Send the given message to the specified exchange. - * + * * @param channel the RabbitMQ Channel to operate within * @param exchange the name of the RabbitMQ exchange to send to * @param routingKey the routing key @@ -657,7 +657,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, /** * Check whether the given Channel is locally transacted, that is, whether its transaction is managed by this * template's Channel handling and not by an external transaction coordinator. - * + * * @param channel the Channel to check * @return whether the given Channel is locally transacted * @see ConnectionFactoryUtils#isChannelTransactional diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index a7253442..52e47954 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -1,11 +1,11 @@ /* * Copyright 2002-2011 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. @@ -81,9 +81,9 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im * {@link AcknowledgeMode#NONE} then the channel cannot be transactional (so the container will fail on start up if * that flag is accidentally set). *
- * + * * @param acknowledgeMode the acknowledge mode to set. Defaults to {@link AcknowledgeMode#AUTO} - * + * * @see AcknowledgeMode */ public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) { @@ -167,7 +167,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im * Check the given message listener, throwing an exception if it does not correspond to a supported listener type. ** By default, only a Spring {@link MessageListener} object or a Spring - * {@link org.springframework.jms.listener.SessionAwareMessageListener} object will be accepted. + * {@link ChannelAwareMessageListener} object will be accepted. * @param messageListener the message listener object to check * @throws IllegalArgumentException if the supplied listener is not a MessageListener or SessionAwareMessageListener * @see MessageListener @@ -238,6 +238,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im /** * Delegates to {@link #validateConfiguration()} and {@link #initialize()}. */ + @Override public final void afterPropertiesSet() { super.afterPropertiesSet(); Assert.state( diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java index d2badfbc..9feb342d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java @@ -1,11 +1,11 @@ /* * 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. @@ -45,80 +45,80 @@ import com.rabbitmq.client.Channel; * Message listener adapter that delegates the handling of messages to target listener methods via reflection, with * flexible message type conversion. Allows listener methods to operate on message content types, completely independent * from the Rabbit API. - * + * *
* By default, the content of incoming Rabbit messages gets extracted before being passed into the target listener
* method, to let the target method operate on message content types such as String or byte array instead of the raw
* {@link Message}. Message type conversion is delegated to a Spring AMQ {@link MessageConverter}. By default, a
* {@link SimpleMessageConverter} will be used. (If you do not want such automatic message conversion taking place, then
* be sure to set the {@link #setMessageConverter MessageConverter} to null.)
- *
+ *
*
* If a target listener method returns a non-null object (typically of a message content type such as
* String or byte array), it will get wrapped in a Rabbit Message and sent to the exchange of
* the incoming message with the routingKey that comes from the Rabbit ReplyTo property or via
* {@link #setResponseRoutingKey(String) specified routingKey}).
- *
+ *
*
* Note: The sending of response messages is only available when using the {@link ChannelAwareMessageListener} * entry point (typically through a Spring message listener container). Usage as {@link MessageListener} does not * support the generation of response messages. - * + * *
* Find below some examples of method signatures compliant with this adapter class. This first example handles all
* Message types and gets passed the contents of each Message type as an argument. No
* Message will be sent back as all of these methods return void.
- *
+ *
*
* public interface MessageContentsDelegate {
* void handleMessage(String text);
- *
+ *
* void handleMessage(Map map);
- *
+ *
* void handleMessage(byte[] bytes);
- *
+ *
* void handleMessage(Serializable obj);
* }
*
- *
+ *
* This next example handle a Message type and gets passed the actual (raw) Message as an
* argument. Again, no Message will be sent back as all of these methods return void.
- *
+ *
*
* public interface RawMessageDelegate {
* void handleMessage(Message message);
* }
*
- *
+ *
* This next example illustrates a Message delegate that just consumes the String contents of
* {@link Message Messages}. Notice also how the name of the Message handling method is different from the
* {@link #ORIGINAL_DEFAULT_LISTENER_METHOD original} (this will have to be configured in the attandant bean
* definition). Again, no Message will be sent back as the method returns void.
- *
+ *
*
* public interface TextMessageContentDelegate {
* void onMessage(String text);
* }
*
- *
+ *
* This final example illustrates a Message delegate that just consumes the String contents of
* {@link Message Messages}. Notice how the return type of this method is String: This will result in the
* configured {@link MessageListenerAdapter} sending a {@link Message} in response.
- *
+ *
*
* public interface ResponsiveTextMessageContentDelegate {
* String handleMessage(String text);
* }
*
- *
+ *
* For further examples and discussion please do refer to the Spring reference documentation which describes this class
* (and its attendant XML configuration) in detail.
- *
+ *
* @author Juergen Hoeller
* @author Mark Pollack
* @author Mark Fisher
* @author Dave Syer
- *
+ *
* @see #setDelegate
* @see #setDefaultListenerMethod
* @see #setResponseRoutingKey(String)
@@ -208,7 +208,7 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
/**
* The encoding to use when inter-converting between byte arrays and Strings in message properties.
- *
+ *
* @param encoding the encoding to set
*/
public void setEncoding(String encoding) {
@@ -362,8 +362,8 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
* Handle the given exception that arose during listener execution. The default implementation logs the exception at
* error level.
* - * This method only applies when using a Rabbit {@link MessageListener}. In case of the Spring - * {@link org.springframework.jms.listener.SessionAwareMessageListener} mechanism, exceptions get handled by the + * This method only applies when using a Rabbit {@link MessageListener}. With + * {@link ChannelAwareMessageListener}, exceptions get handled by the * caller instead. * @param ex the exception to handle * @see #onMessage(Message) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java index 4f87fa8f..d7a6ca82 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java @@ -31,7 +31,7 @@ public class PendingConfirm { private final long timestamp; /** - * @param correlationId + * @param correlationData * @param timestamp */ public PendingConfirm(CorrelationData correlationData, long timestamp) { 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 ef172132..f09cf3d7 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 @@ -1,11 +1,11 @@ /* * Copyright 2002-2011 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. @@ -35,29 +35,29 @@ import com.rabbitmq.client.Connection; * {@link org.springframework.transaction.PlatformTransactionManager} implementation for a single Rabbit * {@link ConnectionFactory}. Binds a Rabbit Channel from the specified ConnectionFactory to the thread, potentially * allowing for one thread-bound channel per ConnectionFactory. - * + * *
* This local strategy is an alternative to executing Rabbit operations within, and synchronized with, external * transactions. This strategy is not able to provide XA transactions, for example in order to share transactions * between messaging and database access. - * + * *
* Application code is required to retrieve the transactional Rabbit resources via * {@link ConnectionFactoryUtils#getTransactionalResourceHolder(ConnectionFactory, boolean)} instead of a standard * {@link Connection#createChannel()} call with subsequent Channel creation. Spring's {@link RabbitTemplate} will * autodetect a thread-bound Channel and automatically participate in it. - * + * *
* The use of {@link CachingConnectionFactory} as a target for this transaction manager is strongly recommended. * CachingConnectionFactory uses a single Rabbit Connection for all Rabbit access in order to avoid the overhead of * repeated Connection creation, as well as maintaining a cache of Channels. Each transaction will then share the same * Rabbit Connection, while still using its own individual Rabbit Channel. - * + * *
* Transaction synchronization is turned off by default, as this manager might be used alongside a datastore-based - * Spring transaction manager such as the JDBC {@link org.springframework.jdbc.datasource.DataSourceTransactionManager}, + * Spring transaction manager such as the JDBC org.springframework.jdbc.datasource.DataSourceTransactionManager, * which has stronger needs for synchronization. - * + * * @author Dave Syer */ @SuppressWarnings("serial") @@ -119,6 +119,7 @@ public class RabbitTransactionManager extends AbstractPlatformTransactionManager return getConnectionFactory(); } + @Override protected Object doGetTransaction() { RabbitTransactionObject txObject = new RabbitTransactionObject(); txObject.setResourceHolder((RabbitResourceHolder) TransactionSynchronizationManager @@ -126,11 +127,13 @@ public class RabbitTransactionManager extends AbstractPlatformTransactionManager return txObject; } + @Override protected boolean isExistingTransaction(Object transaction) { RabbitTransactionObject txObject = (RabbitTransactionObject) transaction; return (txObject.getResourceHolder() != null); } + @Override protected void doBegin(Object transaction, TransactionDefinition definition) { if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) { throw new InvalidIsolationLevelException("AMQP does not support an isolation level concept"); @@ -158,34 +161,40 @@ public class RabbitTransactionManager extends AbstractPlatformTransactionManager } } + @Override protected Object doSuspend(Object transaction) { RabbitTransactionObject txObject = (RabbitTransactionObject) transaction; txObject.setResourceHolder(null); return TransactionSynchronizationManager.unbindResource(getConnectionFactory()); } + @Override protected void doResume(Object transaction, Object suspendedResources) { RabbitResourceHolder conHolder = (RabbitResourceHolder) suspendedResources; TransactionSynchronizationManager.bindResource(getConnectionFactory(), conHolder); } + @Override protected void doCommit(DefaultTransactionStatus status) { RabbitTransactionObject txObject = (RabbitTransactionObject) status.getTransaction(); RabbitResourceHolder resourceHolder = txObject.getResourceHolder(); resourceHolder.commitAll(); } + @Override protected void doRollback(DefaultTransactionStatus status) { RabbitTransactionObject txObject = (RabbitTransactionObject) status.getTransaction(); RabbitResourceHolder resourceHolder = txObject.getResourceHolder(); resourceHolder.rollbackAll(); } + @Override protected void doSetRollbackOnly(DefaultTransactionStatus status) { RabbitTransactionObject txObject = (RabbitTransactionObject) status.getTransaction(); txObject.getResourceHolder().setRollbackOnly(); } + @Override protected void doCleanupAfterCompletion(Object transaction) { RabbitTransactionObject txObject = (RabbitTransactionObject) transaction; TransactionSynchronizationManager.unbindResource(getConnectionFactory());