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

Resolve new tangles

Browse Source 
- `MessagePublishingErrorHandler <-> IntegrationContextUtils`
This commit is contained in:
Gary Russell
2019-04-10 12:16:47 -04:00
committed by Artem Bilan
parent 6c3ffcb1d5
commit a40f20f40d
17 changed files with 146 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java
View File

@@ -31,13 +31,13 @@ import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
import org.springframework.core.ParameterNameDiscoverer;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.expression.ExpressionEvalMap;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -104,7 +104,7 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
this.beanFactory = beanFactory;
this.messagingTemplate.setBeanFactory(beanFactory);
if (this.channelResolver == null) {
this.channelResolver = IntegrationContextUtils.getChannelResolver(this.beanFactory);
this.channelResolver = ChannelResolverUtils.getChannelResolver(this.beanFactory);
}
}

59
spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelUtils.java Normal file
View File

@@ -0,0 +1,59 @@
/*
* Copyright 2019 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
*
* https://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.integration.channel;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
/**
* Channel utilities.
*
* @author Artem Bilan
* @author Gary Russell
* @since 5.2
*
*/
public final class ChannelUtils {
public static final String MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME = "integrationMessagePublishingErrorHandler";
private ChannelUtils() {
super();
}
/**
* Obtain an {@link ErrorHandler} registered with the
* {@value MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME} bean name.
* By default resolves to the {@link org.springframework.integration.channel.MessagePublishingErrorHandler}
* with the {@value ChannelResolverUtils#CHANNEL_RESOLVER_BEAN_NAME} {@link DestinationResolver} bean.
* @param beanFactory BeanFactory for lookup, must not be null.
* @return the instance of {@link ErrorHandler} bean whose name is
* {@value MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME}.
*/
public static ErrorHandler getErrorHandler(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "'beanFactory' must not be null");
if (!beanFactory.containsBean(MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME)) {
return new MessagePublishingErrorHandler(ChannelResolverUtils.getChannelResolver(beanFactory));
}
return beanFactory.getBean(MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME, ErrorHandler.class);
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
View File

@@ -18,7 +18,6 @@ package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
@@ -103,7 +102,7 @@ public class ExecutorChannel extends AbstractExecutorChannel {
+ "bean is fully initialized by the framework. Do not subscribe in a @Bean definition");
super.onInit();
if (!(this.executor instanceof ErrorHandlingTaskExecutor)) {
ErrorHandler errorHandler = IntegrationContextUtils.getErrorHandler(getBeanFactory());
ErrorHandler errorHandler = ChannelUtils.getErrorHandler(getBeanFactory());
this.executor = new ErrorHandlingTaskExecutor(this.executor, errorHandler);
}
UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(this.executor);

3
spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
@@ -139,7 +138,7 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel {
+ "bean is fully initialized by the framework. Do not subscribe in a @Bean definition");
if (!(this.executor instanceof ErrorHandlingTaskExecutor)) {
if (this.errorHandler == null) {
this.errorHandler = IntegrationContextUtils.getErrorHandler(beanFactory);
this.errorHandler = ChannelUtils.getErrorHandler(beanFactory);
}
this.executor = new ErrorHandlingTaskExecutor(this.executor, this.errorHandler);
}

4
spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java
View File

@@ -23,8 +23,8 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.context.Lifecycle;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageSelector;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -182,7 +182,7 @@ public class WireTap implements ChannelInterceptor, Lifecycle, VetoCapableInterc
String channelNameToUse = this.channelName;
if (channelNameToUse != null) {
this.channel =
IntegrationContextUtils.getChannelResolver(this.beanFactory)
ChannelResolverUtils.getChannelResolver(this.beanFactory)
.resolveDestination(channelNameToUse);
this.channelName = null;
}

12
spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java
View File

@@ -49,6 +49,7 @@ import org.springframework.context.Lifecycle;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
import org.springframework.integration.channel.ChannelUtils;
import org.springframework.integration.channel.DefaultHeaderChannelRegistry;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.channel.NullChannel;
@@ -64,6 +65,7 @@ import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.NullAwarePayloadArgumentResolver;
import org.springframework.integration.support.SmartLifecycleRoleController;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
import org.springframework.integration.support.json.JacksonPresent;
@@ -149,16 +151,16 @@ class DefaultConfiguringBeanFactoryPostProcessor
}
private void registerBeanFactoryChannelResolver() {
if (!this.beanFactory.containsBeanDefinition(IntegrationContextUtils.CHANNEL_RESOLVER_BEAN_NAME)) {
this.registry.registerBeanDefinition(IntegrationContextUtils.CHANNEL_RESOLVER_BEAN_NAME,
if (!this.beanFactory.containsBeanDefinition(ChannelResolverUtils.CHANNEL_RESOLVER_BEAN_NAME)) {
this.registry.registerBeanDefinition(ChannelResolverUtils.CHANNEL_RESOLVER_BEAN_NAME,
new RootBeanDefinition(BeanFactoryChannelResolver.class));
}
}
private void registerMessagePublishingErrorHandler() {
if (!this.beanFactory.containsBeanDefinition(
IntegrationContextUtils.MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME)) {
this.registry.registerBeanDefinition(IntegrationContextUtils.MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME,
ChannelUtils.MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME)) {
this.registry.registerBeanDefinition(ChannelUtils.MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME,
new RootBeanDefinition(MessagePublishingErrorHandler.class));
}
}
@@ -301,7 +303,7 @@ class DefaultConfiguringBeanFactoryPostProcessor
.addPropertyValue("threadNamePrefix", "task-scheduler-")
.addPropertyValue("rejectedExecutionHandler", new CallerRunsPolicy())
.addPropertyReference("errorHandler",
IntegrationContextUtils.MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME)
ChannelUtils.MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME)
.getBeanDefinition();
this.registry.registerBeanDefinition(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, scheduler);

4
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -54,7 +54,6 @@ import org.springframework.integration.annotation.Poller;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.config.IntegrationConfigUtils;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.Orderable;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.AbstractPollingEndpoint;
@@ -70,6 +69,7 @@ import org.springframework.integration.handler.ReplyProducingMessageHandlerWrapp
import org.springframework.integration.handler.advice.HandleMessageAdvice;
import org.springframework.integration.router.AbstractMessageRouter;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.integration.util.ClassUtils;
import org.springframework.integration.util.MessagingAnnotationUtils;
import org.springframework.lang.Nullable;
@@ -125,7 +125,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
this.conversionService = this.beanFactory.getConversionService() != null
? this.beanFactory.getConversionService()
: DefaultConversionService.getSharedInstance();
this.channelResolver = IntegrationContextUtils.getChannelResolver(beanFactory);
this.channelResolver = ChannelResolverUtils.getChannelResolver(beanFactory);
this.annotationType =
(Class<T>) GenericTypeResolver.resolveTypeArgument(this.getClass(),
MethodAnnotationPostProcessor.class);

43
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
View File

@@ -23,15 +23,11 @@ import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.expression.spel.support.SimpleEvaluationContext;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.config.IntegrationConfigUtils;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
/**
* Utility methods for accessing common integration components from the BeanFactory.
@@ -111,10 +107,6 @@ public abstract class IntegrationContextUtils {
public static final String LIST_MESSAGE_HANDLER_FACTORY_BEAN_NAME = "integrationListMessageHandlerMethodFactory";
public static final String CHANNEL_RESOLVER_BEAN_NAME = "integrationChannelResolver";
public static final String MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME = "integrationMessagePublishingErrorHandler";
/**
* @param beanFactory BeanFactory for lookup, must not be null.
* @return The {@link MetadataStore} bean whose name is "metadataStore".
@@ -220,39 +212,4 @@ public abstract class IntegrationContextUtils {
return properties;
}
/**
* Obtain a {@link DestinationResolver} registered with the
* {@value #CHANNEL_RESOLVER_BEAN_NAME} bean name.
* @param beanFactory BeanFactory for lookup, must not be null.
* @return the instance of {@link DestinationResolver} bean whose name is
* {@value #CHANNEL_RESOLVER_BEAN_NAME}.
* @since 5.2
*/
@SuppressWarnings("unchecked")
public static DestinationResolver<MessageChannel> getChannelResolver(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "'beanFactory' must not be null");
if (!beanFactory.containsBean(CHANNEL_RESOLVER_BEAN_NAME)) {
return new BeanFactoryChannelResolver(beanFactory);
}
return beanFactory.getBean(CHANNEL_RESOLVER_BEAN_NAME, DestinationResolver.class);
}
/**
* Obtain an {@link ErrorHandler} registered with the
* {@value #MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME} bean name.
* By default resolves to the {@link org.springframework.integration.channel.MessagePublishingErrorHandler}
* with the {@value #CHANNEL_RESOLVER_BEAN_NAME} {@link DestinationResolver} bean.
* @param beanFactory BeanFactory for lookup, must not be null.
* @return the instance of {@link ErrorHandler} bean whose name is
* {@value #MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME}.
* @since 5.2
*/
public static ErrorHandler getErrorHandler(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "'beanFactory' must not be null");
if (!beanFactory.containsBean(MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME)) {
return new MessagePublishingErrorHandler(getChannelResolver(beanFactory));
}
return beanFactory.getBean(MESSAGE_PUBLISHING_ERROR_HANDLER_BEAN_NAME, ErrorHandler.class);
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -37,6 +37,7 @@ import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.lang.Nullable;
@@ -227,7 +228,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
protected DestinationResolver<MessageChannel> getChannelResolver() {
if (this.channelResolver == null) {
this.channelResolver = IntegrationContextUtils.getChannelResolver(this.beanFactory);
this.channelResolver = ChannelResolverUtils.getChannelResolver(this.beanFactory);
}
return this.channelResolver;
}

3
spring-integration-core/src/main/java/org/springframework/integration/core/ErrorMessagePublisher.java
View File

@@ -26,6 +26,7 @@ import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.support.DefaultErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -96,7 +97,7 @@ public class ErrorMessagePublisher implements BeanFactoryAware {
public void setBeanFactory(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "beanFactory must not be null");
if (this.channelResolver == null) {
this.channelResolver = IntegrationContextUtils.getChannelResolver(beanFactory);
this.channelResolver = ChannelResolverUtils.getChannelResolver(beanFactory);
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java
View File

@@ -22,6 +22,7 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.GenericMessagingTemplate;
@@ -61,7 +62,7 @@ public class MessagingTemplate extends GenericMessagingTemplate {
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory; //NOSONAR - non-sync is ok here
setDestinationResolver(IntegrationContextUtils.getChannelResolver(beanFactory));
setDestinationResolver(ChannelResolverUtils.getChannelResolver(beanFactory));
}
@Override

4
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -32,8 +32,8 @@ import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.integration.channel.ChannelUtils;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.support.MessagingExceptionWrapper;
import org.springframework.integration.transaction.IntegrationResourceHolder;
import org.springframework.integration.transaction.IntegrationResourceHolderSynchronization;
@@ -196,7 +196,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
if (this.taskExecutor != null) {
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
if (this.errorHandler == null) {
this.errorHandler = IntegrationContextUtils.getErrorHandler(getBeanFactory());
this.errorHandler = ChannelUtils.getErrorHandler(getBeanFactory());
this.errorHandlerIsDefault = true;
}
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, this.errorHandler);

4
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
View File

@@ -23,9 +23,9 @@ import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelUtils;
import org.springframework.integration.channel.MessageChannelReactiveUtils;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.router.MessageRouter;
import org.springframework.messaging.Message;
@@ -123,7 +123,7 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
protected void onInit() {
super.onInit();
if (this.errorHandler == null) {
this.errorHandler = IntegrationContextUtils.getErrorHandler(getBeanFactory());
this.errorHandler = ChannelUtils.getErrorHandler(getBeanFactory());
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -55,11 +55,11 @@ import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.GatewayHeader;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.integration.support.management.TrackableComponent;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
@@ -384,7 +384,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
}
BeanFactory beanFactory = this.getBeanFactory();
if (this.channelResolver == null && beanFactory != null) {
this.channelResolver = IntegrationContextUtils.getChannelResolver(beanFactory);
this.channelResolver = ChannelResolverUtils.getChannelResolver(beanFactory);
}
Class<?> proxyInterface = determineServiceInterface();
Method[] methods = ReflectionUtils.getUniqueDeclaredMethods(proxyInterface);

56
spring-integration-core/src/main/java/org/springframework/integration/support/channel/ChannelResolverUtils.java Normal file
View File

@@ -0,0 +1,56 @@
/*
* Copyright 2019 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
*
* https://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.integration.support.channel;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.util.Assert;
/**
* Channel resolution utilities.
*
* @author Artem Bilan
* @author Gary Russell
* @since 5.2
*
*/
public final class ChannelResolverUtils {
public static final String CHANNEL_RESOLVER_BEAN_NAME = "integrationChannelResolver";
private ChannelResolverUtils() {
super();
}
/**
* Obtain a {@link DestinationResolver} registered with the
* {@value CHANNEL_RESOLVER_BEAN_NAME} bean name.
* @param beanFactory BeanFactory for lookup, must not be null.
* @return the instance of {@link DestinationResolver} bean whose name is
* {@value CHANNEL_RESOLVER_BEAN_NAME}.
*/
@SuppressWarnings("unchecked")
public static DestinationResolver<MessageChannel> getChannelResolver(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "'beanFactory' must not be null");
if (!beanFactory.containsBean(CHANNEL_RESOLVER_BEAN_NAME)) {
return new BeanFactoryChannelResolver(beanFactory);
}
return beanFactory.getBean(CHANNEL_RESOLVER_BEAN_NAME, DestinationResolver.class);
}
}

4
spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
View File

@@ -33,7 +33,7 @@ import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.channel.ChannelUtils;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.support.converter.SimpleMessageConverter;
@@ -149,7 +149,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
}
this.container.setConnectionFactory(this.connectionFactory);
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
ErrorHandler errorHandler = IntegrationContextUtils.getErrorHandler(beanFactory);
ErrorHandler errorHandler = ChannelUtils.getErrorHandler(beanFactory);
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler);
}
this.container.setTaskExecutor(this.taskExecutor);

4
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java
View File

@@ -31,9 +31,9 @@ import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.redis.event.RedisExceptionEvent;
import org.springframework.integration.support.channel.ChannelResolverUtils;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.jmx.export.annotation.ManagedMetric;
@@ -177,7 +177,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
BeanFactory beanFactory = getBeanFactory();
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && beanFactory != null) {
MessagePublishingErrorHandler errorHandler =
new MessagePublishingErrorHandler(IntegrationContextUtils.getChannelResolver(beanFactory));
new MessagePublishingErrorHandler(ChannelResolverUtils.getChannelResolver(beanFactory));
errorHandler.setDefaultErrorChannel(this.errorChannel);
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler);
}