From 344fcb9f035e8d53f2579adb82756f1b25c72a2d Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 28 Jun 2015 12:42:18 -0400 Subject: [PATCH] AMQP-508: Queue Declaration IllegalArgumentExcept. JIRA: https://jira.spring.io/browse/AMQP-508 Workaround for https://github.com/rabbitmq/rabbitmq-java-client/issues/72 --- .../amqp/rabbit/core/RabbitAdmin.java | 30 +++++++++++++++++-- .../listener/BlockingQueueConsumer.java | 15 +++++++++- .../amqp/rabbit/core/RabbitAdminTests.java | 14 +++++++-- ...BlockingQueueConsumerIntegrationTests.java | 21 +++++++++++++ 4 files changed, 74 insertions(+), 6 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index a91571e7..95e18a27 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java @@ -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) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 56c4c867..aca63232 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -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()) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java index 3604f71a..b576b3d3 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java @@ -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); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java index 70307a3a..9118e10c 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java @@ -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(), + AcknowledgeMode.AUTO, true, 1, longName, "foobar"); + try { + blockingQueueConsumer.start(); + fail("expected exception"); + } + catch (FatalListenerStartupException e) { + assertThat(e.getCause(), instanceOf(IllegalArgumentException.class)); + } + connectionFactory.destroy(); + } + }