GH-723: Add AmqpResourceNotAvailableException (#737)
* GH-723: Add AmqpResourceNotAvailableException Fixes spring-projects/spring-amqp#723 To avoid an `NPE` when connection returns `null` for the `createChannel()` in case of `channelMax` is reached, throw newly introduced `AmqpResourceNotAvailableException`. This exception can be used in the `RetryPolicy` to retry the original operation after some back-off - the channel permit may be released in between **Cherry-pick to 2.0.x and 1.7.x** * * Fix `AbstractConnectionFactoryTests` for proper mock * Fix `amqp.adoc` according PR comments
This commit is contained in:
committed by
Gary Russell
parent
6498219274
commit
f69f54e828
@@ -0,0 +1,42 @@
|
||||
/*
|
||||
* Copyright 2018 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.amqp;
|
||||
|
||||
/**
|
||||
* The {@link AmqpException} thrown when some resource can't be accessed.
|
||||
* For example when {@code channelMax} limit is reached and connect can't
|
||||
* create a new channel at the moment.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 1.7.7
|
||||
*/
|
||||
public class AmqpResourceNotAvailableException extends AmqpException {
|
||||
|
||||
public AmqpResourceNotAvailableException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
public AmqpResourceNotAvailableException(Throwable cause) {
|
||||
super(cause);
|
||||
}
|
||||
|
||||
public AmqpResourceNotAvailableException(String message, Throwable cause) {
|
||||
super(message, cause);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2017 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.connection;
|
||||
import java.io.IOException;
|
||||
import java.net.InetAddress;
|
||||
|
||||
import org.springframework.amqp.AmqpResourceNotAvailableException;
|
||||
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
|
||||
@@ -53,6 +54,9 @@ public class SimpleConnection implements Connection, NetworkConnection {
|
||||
public Channel createChannel(boolean transactional) {
|
||||
try {
|
||||
Channel channel = this.delegate.createChannel();
|
||||
if (channel == null) {
|
||||
throw new AmqpResourceNotAvailableException("The channelMax limit is reached. Try later.");
|
||||
}
|
||||
if (transactional) {
|
||||
// Just created so we want to start the transaction
|
||||
channel.txSelect();
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2010-2016 the original author or authors.
|
||||
* Copyright 2010-2018 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.
|
||||
@@ -46,6 +46,7 @@ import org.springframework.amqp.utils.test.TestUtils;
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
|
||||
|
||||
import com.rabbitmq.client.Channel;
|
||||
import com.rabbitmq.client.ConnectionFactory;
|
||||
|
||||
/**
|
||||
@@ -176,6 +177,7 @@ public abstract class AbstractConnectionFactoryTests {
|
||||
.thenReturn(mockConnection1, mockConnection2);
|
||||
// simulate a dead connection
|
||||
when(mockConnection1.isOpen()).thenReturn(false);
|
||||
when(mockConnection2.createChannel()).thenReturn(mock(Channel.class));
|
||||
|
||||
AbstractConnectionFactory connectionFactory = createConnectionFactory(mockConnectionFactory);
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2017 the original author or authors.
|
||||
* Copyright 2002-2018 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.
|
||||
@@ -60,6 +60,7 @@ import org.junit.rules.ExpectedException;
|
||||
import org.springframework.amqp.AmqpAuthenticationException;
|
||||
import org.springframework.amqp.AmqpException;
|
||||
import org.springframework.amqp.AmqpIOException;
|
||||
import org.springframework.amqp.AmqpResourceNotAvailableException;
|
||||
import org.springframework.amqp.AmqpTimeoutException;
|
||||
import org.springframework.amqp.core.Queue;
|
||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
|
||||
@@ -87,6 +88,7 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
|
||||
* @author Gunnar Hillert
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 1.0
|
||||
*
|
||||
*/
|
||||
@@ -568,6 +570,14 @@ public class CachingConnectionFactoryIntegrationTests {
|
||||
factory.destroy();
|
||||
}
|
||||
|
||||
@Test(expected = AmqpResourceNotAvailableException.class)
|
||||
public void testChannelMax() {
|
||||
this.connectionFactory.getRabbitConnectionFactory().setRequestedChannelMax(1);
|
||||
Connection connection = this.connectionFactory.createConnection();
|
||||
connection.createChannel(true);
|
||||
connection.createChannel(false);
|
||||
}
|
||||
|
||||
private Log spyOnLogger(CachingConnectionFactory connectionFactory2) {
|
||||
DirectFieldAccessor dfa = new DirectFieldAccessor(connectionFactory2);
|
||||
Log logger = spy((Log) dfa.getPropertyValue("logger"));
|
||||
|
||||
@@ -361,6 +361,9 @@ The `ConnectionFactory` argument can be used to distinguish target connection na
|
||||
By default a `beanName` of the `AbstractConnectionFactory` and an internal counter are used to generate `connection_name`.
|
||||
The `<rabbit:connection-factory>` namespace component is also supplied with the `connection-name-strategy` attribute.
|
||||
|
||||
Starting with _version 1.7.7_, an `AmqpResourceNotAvailableException` is provided, which is thrown now when `SimpleConnection.createChannel()` can't create a `Channel`, for example, because the `channelMax` limit is reached and there are no available channels in the cache.
|
||||
This exception can be used in the `RetryPolicy` to recover the operation after some back-off.
|
||||
|
||||
[[connection-factory]]
|
||||
===== Configuring the Underlying Client Connection Factory
|
||||
|
||||
|
||||
Reference in New Issue
Block a user