From 7abefb08dfb0bc1de3d018e0d619a45f1c7c1964 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 17 Jul 2014 18:13:06 +0300 Subject: [PATCH] INT-3475: `Integer.MAX_VALUE / 2` Phase for MDEs JIRA: https://jira.spring.io/browse/INT-3475 Change the phase for `SourcePollingChannelAdapterFactoryBean`, `AbstractPollingEndpoint`, `MessageProducerSupport`, `JmsMessageDrivenEndpoint` to the `Integer.MAX_VALUE / 2` --- .../AmqpInboundChannelAdapterParserTests.java | 2 +- .../config/ConsumerEndpointFactoryBean.java | 21 +++++++++++++---- ...ourcePollingChannelAdapterFactoryBean.java | 7 +++--- .../endpoint/AbstractPollingEndpoint.java | 2 +- .../endpoint/MessageProducerSupport.java | 3 +++ ...licationEventListeningMessageProducer.java | 20 +++++----------- .../jms/JmsMessageDrivenEndpoint.java | 9 ++++---- ...QueueInboundChannelAdapterParserTests.java | 23 +++++++++++-------- 8 files changed, 50 insertions(+), 37 deletions(-) diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java index 525b9581c4..ed132817d2 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java @@ -62,7 +62,7 @@ public class AmqpInboundChannelAdapterParserTests { assertEquals(DirectChannel.class, channel.getClass()); assertEquals(AmqpInboundChannelAdapter.class, adapter.getClass()); assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(adapter, "autoStartup")); - assertEquals(0, TestUtils.getPropertyValue(adapter, "phase")); + assertEquals(Integer.MAX_VALUE / 2, TestUtils.getPropertyValue(adapter, "phase")); assertTrue(TestUtils.getPropertyValue(adapter, "messageListenerContainer.missingQueuesFatal", Boolean.class)); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 1c65641677..225103fe9f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -54,9 +54,11 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky * @author Josh Long * @author Gary Russell + * @author Artem Bilan */ public class ConsumerEndpointFactoryBean - implements FactoryBean, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, SmartLifecycle { + implements FactoryBean, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, + InitializingBean, SmartLifecycle { private volatile MessageHandler handler; @@ -70,6 +72,8 @@ public class ConsumerEndpointFactoryBean private volatile int phase = 0; + private volatile boolean isPhaseSet; + private volatile MessageChannel inputChannel; private volatile ConfigurableBeanFactory beanFactory; @@ -119,6 +123,7 @@ public class ConsumerEndpointFactoryBean public void setPhase(int phase) { this.phase = phase; + this.isPhaseSet = true; } @Override @@ -246,7 +251,8 @@ public class ConsumerEndpointFactoryBean pollingConsumer.setErrorHandler(this.pollerMetadata.getErrorHandler()); pollingConsumer.setReceiveTimeout(this.pollerMetadata.getReceiveTimeout()); - pollingConsumer.setTransactionSynchronizationFactory(this.pollerMetadata.getTransactionSynchronizationFactory()); + pollingConsumer.setTransactionSynchronizationFactory( + this.pollerMetadata.getTransactionSynchronizationFactory()); pollingConsumer.setBeanClassLoader(beanClassLoader); pollingConsumer.setBeanFactory(beanFactory); this.endpoint = pollingConsumer; @@ -257,7 +263,11 @@ public class ConsumerEndpointFactoryBean this.endpoint.setBeanName(this.beanName); this.endpoint.setBeanFactory(this.beanFactory); this.endpoint.setAutoStartup(this.autoStartup); - this.endpoint.setPhase(this.phase); + int phase = this.phase; + if (!this.isPhaseSet && this.endpoint instanceof PollingConsumer) { + phase = Integer.MAX_VALUE / 2; + } + this.endpoint.setPhase(phase); this.endpoint.afterPropertiesSet(); this.initialized = true; } @@ -270,7 +280,7 @@ public class ConsumerEndpointFactoryBean @Override public boolean isAutoStartup() { - return (this.endpoint != null) ? this.endpoint.isAutoStartup() : true; + return (this.endpoint == null) || this.endpoint.isAutoStartup(); } @Override @@ -280,7 +290,7 @@ public class ConsumerEndpointFactoryBean @Override public boolean isRunning() { - return (this.endpoint != null) ? this.endpoint.isRunning() : false; + return (this.endpoint != null) && this.endpoint.isRunning(); } @Override @@ -303,4 +313,5 @@ public class ConsumerEndpointFactoryBean this.endpoint.stop(callback); } } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java index 1eaf5d6aaf..1f6a01a86c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java @@ -53,7 +53,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean> eventTypes; @@ -55,8 +56,6 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP private volatile long stoppedAt; - private volatile boolean phaseSet; - /** * Set the list of event types (classes that extend ApplicationEvent) that * this adapter should send to the message channel. By default, all event @@ -81,12 +80,6 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP } } - @Override - public void setPhase(int phase) { - super.setPhase(phase); - this.phaseSet = true; - } - @Override public String getComponentType() { return "event:inbound-channel-adapter"; @@ -96,12 +89,11 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP protected void onInit() { super.onInit(); this.applicationEventMulticaster = this.getBeanFactory() - .getBean(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME, ApplicationEventMulticaster.class); + .getBean(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME, + ApplicationEventMulticaster.class); Assert.notNull(this.applicationEventMulticaster, - "To use ApplicationListeners the 'applicationEventMulticaster' bean must be supplied within ApplicationContext."); - if (!this.phaseSet) { - super.setPhase(Integer.MIN_VALUE + 1000); - } + "To use ApplicationListeners the 'applicationEventMulticaster' " + + "bean must be supplied within ApplicationContext."); } @Override diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java index baea2ef768..00108de53f 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2014 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. @@ -30,15 +30,15 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @author Gary Russell */ -public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements - DisposableBean, OrderlyShutdownCapable { +public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements DisposableBean, OrderlyShutdownCapable { private final AbstractMessageListenerContainer listenerContainer; private final ChannelPublishingJmsMessageListener listener; - public JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer, ChannelPublishingJmsMessageListener listener) { + public JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer, + ChannelPublishingJmsMessageListener listener) { Assert.notNull(listenerContainer, "listener container must not be null"); Assert.notNull(listener, "listener must not be null"); if (logger.isWarnEnabled() && listenerContainer.getMessageListener() != null) { @@ -48,6 +48,7 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements listenerContainer.setMessageListener(listener); this.listener = listener; this.listenerContainer = listenerContainer; + setPhase(Integer.MAX_VALUE / 2); } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java index c8e8358c16..87fdb6d255 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java @@ -86,30 +86,35 @@ public class RedisQueueInboundChannelAdapterParserTests { @Test public void testInt3017DefaultConfig() { - assertSame(this.connectionFactory, TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.ops.template.connectionFactory")); + assertSame(this.connectionFactory, + TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.ops.template.connectionFactory")); assertEquals("si.test.Int3017.Inbound1", TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.key")); assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "expectMessage", Boolean.class)); - assertEquals(new Long(1000), TestUtils.getPropertyValue(this.defaultAdapter, "receiveTimeout", Long.class)); - assertEquals(new Long(5000), TestUtils.getPropertyValue(this.defaultAdapter, "recoveryInterval", Long.class)); + assertEquals(1000L, TestUtils.getPropertyValue(this.defaultAdapter, "receiveTimeout")); + assertEquals(5000L, TestUtils.getPropertyValue(this.defaultAdapter, "recoveryInterval")); assertNull(TestUtils.getPropertyValue(this.defaultAdapter, "errorChannel")); - assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "taskExecutor"), Matchers.instanceOf(ErrorHandlingTaskExecutor.class)); - assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "serializer"), Matchers.instanceOf(JdkSerializationRedisSerializer.class)); + assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "taskExecutor"), + Matchers.instanceOf(ErrorHandlingTaskExecutor.class)); + assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "serializer"), + Matchers.instanceOf(JdkSerializationRedisSerializer.class)); assertTrue(TestUtils.getPropertyValue(this.defaultAdapter, "autoStartup", Boolean.class)); + assertEquals(Integer.MAX_VALUE / 2, TestUtils.getPropertyValue(this.defaultAdapter, "phase")); assertSame(this.defaultAdapterChannel, TestUtils.getPropertyValue(this.defaultAdapter, "outputChannel")); } @Test public void testInt3017CustomConfig() { - assertSame(this.customRedisConnectionFactory, TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.ops.template.connectionFactory")); + assertSame(this.customRedisConnectionFactory, + TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.ops.template.connectionFactory")); assertEquals("si.test.Int3017.Inbound2", TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.key")); assertTrue(TestUtils.getPropertyValue(this.customAdapter, "expectMessage", Boolean.class)); - assertEquals(new Long(2000), TestUtils.getPropertyValue(this.customAdapter, "receiveTimeout", Long.class)); - assertEquals(new Long(3000), TestUtils.getPropertyValue(this.customAdapter, "recoveryInterval", Long.class)); + assertEquals(2000L, TestUtils.getPropertyValue(this.customAdapter, "receiveTimeout")); + assertEquals(3000L, TestUtils.getPropertyValue(this.customAdapter, "recoveryInterval")); assertSame(this.errorChannel, TestUtils.getPropertyValue(this.customAdapter, "errorChannel")); assertSame(this.taskExecutor, TestUtils.getPropertyValue(this.customAdapter, "taskExecutor")); assertSame(this.serializer, TestUtils.getPropertyValue(this.customAdapter, "serializer")); assertFalse(TestUtils.getPropertyValue(this.customAdapter, "autoStartup", Boolean.class)); - assertEquals(new Integer(100), TestUtils.getPropertyValue(this.customAdapter, "phase", Integer.class)); + assertEquals(100, TestUtils.getPropertyValue(this.customAdapter, "phase")); assertSame(this.sendChannel, TestUtils.getPropertyValue(this.customAdapter, "outputChannel")); }