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

AMQP-508: Queue Declaration IllegalArgumentExcept.

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-508

Workaround for https://github.com/rabbitmq/rabbitmq-java-client/issues/72
This commit is contained in:
Gary Russell
2015-06-28 12:42:18 -04:00
committed by Artem Bilan
parent 460a74a800
commit 344fcb9f03
4 changed files with 74 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
View File

@@ -17,6 +17,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Properties;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
@@ -29,6 +30,7 @@ import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
import org.springframework.amqp.rabbit.connection.ChannelProxy;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
@@ -262,6 +264,16 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
props.put(QUEUE_CONSUMER_COUNT, declareOk.getConsumerCount());
return props;
}
catch (IllegalArgumentException e) {
try {
if (channel instanceof ChannelProxy) {
((ChannelProxy) channel).getTargetChannel().close();
}
}
catch (TimeoutException e1) {
}
return null;
}
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug("Queue '" + queueName + "' does not exist");
@@ -445,9 +457,21 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
logger.debug("declaring Queue '" + queue.getName() + "'");
}
try {
DeclareOk declareOk = channel.queueDeclare(queue.getName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete(),
queue.getArguments());
declareOks[i] = declareOk;
try {
DeclareOk declareOk = channel.queueDeclare(queue.getName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete(),
queue.getArguments());
declareOks[i] = declareOk;
}
catch (IllegalArgumentException e) {
try {
if (channel instanceof ChannelProxy) {
((ChannelProxy) channel).getTargetChannel().close();
}
}
catch (TimeoutException e1) {
}
throw new IOException(e);
}
}
catch (IOException e) {
if (this.ignoreDeclarationExceptions) {

15
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

@@ -40,6 +40,7 @@ import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.ChannelProxy;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
@@ -524,7 +525,19 @@ public class BlockingQueueConsumer {
DeclarationException failures = null;
for (String queueName : this.queues) {
try {
this.channel.queueDeclarePassive(queueName);
try {
this.channel.queueDeclarePassive(queueName);
}
catch (IllegalArgumentException e) {
try {
if (this.channel instanceof ChannelProxy) {
((ChannelProxy) this.channel).getTargetChannel().close();
}
}
catch (TimeoutException e1) {
}
throw new FatalListenerStartupException("Illegal Argument on Queue Declaration", e);
}
}
catch (IOException e) {
if (logger.isWarnEnabled()) {

14
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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
@@ -25,6 +25,7 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -39,15 +40,24 @@ import org.junit.rules.ExpectedException;
import org.mockito.ArgumentCaptor;
import org.mockito.internal.stubbing.answers.DoesNothing;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Binding.DestinationType;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Declarable;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.test.BrokerRunning;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.GenericApplicationContext;
import com.rabbitmq.client.Channel;
@@ -62,7 +72,7 @@ public class RabbitAdminTests {
public BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
@Test
public void testSettingOfNullConectionFactory() {
public void testSettingOfNullConnectionFactory() {
ConnectionFactory connectionFactory = null;
try {
new RabbitAdmin(connectionFactory);

21
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java
View File

@@ -12,8 +12,11 @@
*/
package org.springframework.amqp.rabbit.listener;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import java.util.Map;
import java.util.UUID;
@@ -26,6 +29,7 @@ import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.test.BrokerRunning;
import org.springframework.amqp.rabbit.test.BrokerTestUtils;
@@ -88,4 +92,21 @@ public class BlockingQueueConsumerIntegrationTests {
}
@Test
public void testAvoidHangAMQP_508() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
String longName = new String(new byte[300]).replace('\u0000', 'x');
BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory,
new DefaultMessagePropertiesConverter(), new ActiveObjectCounter<BlockingQueueConsumer>(),
AcknowledgeMode.AUTO, true, 1, longName, "foobar");
try {
blockingQueueConsumer.start();
fail("expected exception");
}
catch (FatalListenerStartupException e) {
assertThat(e.getCause(), instanceOf(IllegalArgumentException.class));
}
connectionFactory.destroy();
}
}