From e13aa288b67f907a87fb4546c3ae503a0c3a95d9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 11 Mar 2019 14:09:30 -0400 Subject: [PATCH] Fix smells for AnnGatePFB & MessagePubErrHandler * Move `Throwable` handling in the `MessagePublishingErrorHandler` to the `handleDeliveryError()` method --- .../MessagePublishingErrorHandler.java | 92 +++++++------- .../AnnotationGatewayProxyFactoryBean.java | 115 ++++++++---------- 2 files changed, 102 insertions(+), 105 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java index 14e173bfdb..66b3b32cb6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-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. @@ -28,6 +28,7 @@ import org.springframework.messaging.core.DestinationResolver; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; +import org.springframework.util.StringUtils; /** * {@link ErrorHandler} implementation that sends an {@link ErrorMessage} to a @@ -43,12 +44,12 @@ public class MessagePublishingErrorHandler extends ErrorMessagePublisher impleme private static final int DEFAULT_SEND_TIMEOUT = 1000; - private static final ErrorMessageStrategy DEFAULT_ERROR_MESSAGE_STRATEGY = (t, a) -> { - if (t instanceof MessagingExceptionWrapper) { - return new ErrorMessage(t.getCause(), ((MessagingExceptionWrapper) t).getFailedMessage()); + private static final ErrorMessageStrategy DEFAULT_ERROR_MESSAGE_STRATEGY = (ex, attrs) -> { + if (ex instanceof MessagingExceptionWrapper) { + return new ErrorMessage(ex.getCause(), ((MessagingExceptionWrapper) ex).getFailedMessage()); } else { - return new ErrorMessage(t); + return new ErrorMessage(ex); } }; @@ -87,65 +88,68 @@ public class MessagePublishingErrorHandler extends ErrorMessagePublisher impleme } @Override - public final void handleError(Throwable t) { - MessageChannel errorChannel = resolveErrorChannel(t); + public final void handleError(Throwable ex) { + MessageChannel errorChannel = resolveErrorChannel(ex); boolean sent = false; if (errorChannel != null) { try { - getMessagingTemplate().send(errorChannel, getErrorMessageStrategy().buildErrorMessage(t, null)); + getMessagingTemplate().send(errorChannel, getErrorMessageStrategy().buildErrorMessage(ex, null)); sent = true; } - catch (Throwable errorDeliveryError) { //NOSONAR - // message will be logged only - if (this.logger.isWarnEnabled()) { - this.logger.warn("Error message was not delivered.", errorDeliveryError); - } - if (errorDeliveryError instanceof Error) { - throw ((Error) errorDeliveryError); - } + catch (Throwable errorDeliveryError) { + handleDeliveryError(errorDeliveryError); } } if (!sent && this.logger.isErrorEnabled()) { - Message failedMessage = (t instanceof MessagingException) ? - ((MessagingException) t).getFailedMessage() : null; - if (failedMessage != null) { - this.logger.error("failure occurred in messaging task with message: " + failedMessage, t); - } - else { - this.logger.error("failure occurred in messaging task", t); - } + Message failedMessage = + ex instanceof MessagingException + ? ((MessagingException) ex).getFailedMessage() + : null; + + this.logger.error("failure occurred in messaging task" + + (failedMessage != null ? " with message: " + failedMessage : ""), ex); + } + } + + private void handleDeliveryError(Throwable errorDeliveryError) { + // message will be logged only + if (this.logger.isWarnEnabled()) { + this.logger.warn("Error message was not delivered.", errorDeliveryError); + } + if (errorDeliveryError instanceof Error) { + throw ((Error) errorDeliveryError); } } @Nullable private MessageChannel resolveErrorChannel(Throwable t) { + DestinationResolver channelResolver = getChannelResolver(); Throwable actualThrowable = t; if (t instanceof MessagingExceptionWrapper) { actualThrowable = t.getCause(); } - Message failedMessage = (actualThrowable instanceof MessagingException) ? - ((MessagingException) actualThrowable).getFailedMessage() : null; - if (getDefaultErrorChannel() == null && getChannelResolver() != null) { - setChannel(getChannelResolver().resolveDestination(// NOSONAR not null - IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)); + Message failedMessage = + actualThrowable instanceof MessagingException + ? ((MessagingException) actualThrowable).getFailedMessage() + : null; + if (getDefaultErrorChannel() == null && channelResolver != null) { + setChannel(channelResolver.resolveDestination(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)); } - if (failedMessage == null || failedMessage.getHeaders().getErrorChannel() == null) { - return getDefaultErrorChannel(); - } - Object errorChannelHeader = failedMessage.getHeaders().getErrorChannel(); - if (errorChannelHeader instanceof MessageChannel) { - return (MessageChannel) errorChannelHeader; - } - Assert.isInstanceOf(String.class, errorChannelHeader, () -> - "Unsupported error channel header type. Expected MessageChannel or String, but actual type is [" + - errorChannelHeader.getClass() + "]"); // NOSONAR never null here - if (getChannelResolver() != null) { - return getChannelResolver().resolveDestination((String) errorChannelHeader); // NOSONAR not null - } - else { - return null; + if (failedMessage != null && failedMessage.getHeaders().getErrorChannel() != null) { + Object errorChannelHeader = failedMessage.getHeaders().getErrorChannel(); + if (errorChannelHeader instanceof MessageChannel) { + return (MessageChannel) errorChannelHeader; + } + Assert.isInstanceOf(String.class, errorChannelHeader, () -> + "Unsupported error channel header type. Expected MessageChannel or String, but actual type is [" + + errorChannelHeader.getClass() + "]"); + if (channelResolver != null && StringUtils.hasText((String) errorChannelHeader)) { + return channelResolver.resolveDestination((String) errorChannelHeader); + } } + + return getDefaultErrorChannel(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AnnotationGatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AnnotationGatewayProxyFactoryBean.java index ef106522c3..60ec9bae18 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AnnotationGatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AnnotationGatewayProxyFactoryBean.java @@ -16,9 +16,10 @@ package org.springframework.integration.gateway; -import java.util.HashMap; +import java.util.Arrays; import java.util.Map; import java.util.concurrent.Executor; +import java.util.stream.Collectors; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.core.annotation.AnnotatedElementUtils; @@ -28,6 +29,8 @@ import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.annotation.AnnotationConstants; import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.util.JavaUtils; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -48,6 +51,7 @@ public class AnnotationGatewayProxyFactoryBean extends GatewayProxyFactoryBean { public AnnotationGatewayProxyFactoryBean(Class serviceInterface) { super(serviceInterface); + AnnotationAttributes annotationAttributes = AnnotatedElementUtils.getMergedAnnotationAttributes(serviceInterface, MessagingGateway.class.getName(), false, true); @@ -68,14 +72,37 @@ public class AnnotationGatewayProxyFactoryBean extends GatewayProxyFactoryBean { protected void onInit() { ConfigurableListableBeanFactory beanFactory = (ConfigurableListableBeanFactory) getBeanFactory(); - String defaultPayloadExpression = - beanFactory.resolveEmbeddedValue( - this.gatewayAttributes.getString("defaultPayloadExpression")); + populateGatewayMethodMetadata(); + + JavaUtils.INSTANCE + .acceptIfHasText(resolveAttribute("defaultRequestChannel"), this::setDefaultRequestChannelName) + .acceptIfHasText(resolveAttribute("defaultReplyChannel"), this::setDefaultReplyChannelName) + .acceptIfHasText(resolveAttribute("errorChannel"), this::setErrorChannelName) + .acceptIfHasText(resolveAttribute("defaultRequestTimeout"), + value -> setDefaultRequestTimeout(Long.parseLong(value))) + .acceptIfHasText(resolveAttribute("defaultReplyTimeout"), + value -> setDefaultReplyTimeout(Long.parseLong(value))); + + String asyncExecutor = beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("asyncExecutor")); + if (asyncExecutor == null || AnnotationConstants.NULL.equals(asyncExecutor)) { + setAsyncExecutor(null); + } + else if (StringUtils.hasText(asyncExecutor)) { + setAsyncExecutor(beanFactory.getBean(asyncExecutor, Executor.class)); + } + + super.onInit(); + } + + private void populateGatewayMethodMetadata() { + ConfigurableListableBeanFactory beanFactory = (ConfigurableListableBeanFactory) getBeanFactory(); + + String defaultPayloadExpression = resolveAttribute("defaultPayloadExpression"); @SuppressWarnings("unchecked") Map[] defaultHeaders = (Map[]) this.gatewayAttributes.get("defaultHeaders"); - String mapper = beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("mapper")); + String mapper = resolveAttribute("mapper"); boolean hasMapper = StringUtils.hasText(mapper); boolean hasDefaultPayloadExpression = StringUtils.hasText(defaultPayloadExpression); @@ -86,31 +113,9 @@ public class AnnotationGatewayProxyFactoryBean extends GatewayProxyFactoryBean { Assert.state(!hasMapper || !hasDefaultHeaders, "'defaultHeaders' are not allowed when a 'mapper' is provided"); - - String defaultRequestChannel = - beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("defaultRequestChannel")); - if (StringUtils.hasText(defaultRequestChannel)) { - setDefaultRequestChannelName(defaultRequestChannel); - } - - String defaultReplyChannel = - beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("defaultReplyChannel")); - if (StringUtils.hasText(defaultReplyChannel)) { - setDefaultReplyChannelName(defaultReplyChannel); - } - - String errorChannel = beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("errorChannel")); - if (StringUtils.hasText(errorChannel)) { - setErrorChannelName(errorChannel); - } - - String asyncExecutor = beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("asyncExecutor")); - if (asyncExecutor == null || AnnotationConstants.NULL.equals(asyncExecutor)) { - setAsyncExecutor(null); - } - else if (StringUtils.hasText(asyncExecutor)) { - setAsyncExecutor(beanFactory.getBean(asyncExecutor, Executor.class)); - } + JavaUtils.INSTANCE + .acceptIfHasText(mapper, + value -> setMapper(beanFactory.getBean(value, MethodArgsMessageMapper.class))); if (hasDefaultHeaders || hasDefaultPayloadExpression) { GatewayMethodMetadata gatewayMethodMetadata = new GatewayMethodMetadata(); @@ -119,46 +124,34 @@ public class AnnotationGatewayProxyFactoryBean extends GatewayProxyFactoryBean { gatewayMethodMetadata.setPayloadExpression(defaultPayloadExpression); } - Map headerExpressions = new HashMap<>(); - for (Map header : defaultHeaders) { - String headerValue = beanFactory.resolveEmbeddedValue((String) header.get("value")); - boolean hasValue = StringUtils.hasText(headerValue); + Map headerExpressions = Arrays.stream(defaultHeaders) + .collect(Collectors.toMap( + header -> beanFactory.resolveEmbeddedValue((String) header.get("name")), + header -> { + String headerValue = beanFactory.resolveEmbeddedValue((String) header.get("value")); + boolean hasValue = StringUtils.hasText(headerValue); - String headerExpression = beanFactory.resolveEmbeddedValue((String) header.get("expression")); + String headerExpression = + beanFactory.resolveEmbeddedValue((String) header.get("expression")); - Assert.state(!(hasValue == StringUtils.hasText(headerExpression)), - "exactly one of 'value' or 'expression' is required on a gateway's header."); + Assert.state(!(hasValue == StringUtils.hasText(headerExpression)), + "exactly one of 'value' or 'expression' is required on a gateway's header."); - Expression expression = hasValue ? - new LiteralExpression(headerValue) : - EXPRESSION_PARSER.parseExpression(headerExpression); - - String headerName = beanFactory.resolveEmbeddedValue((String) header.get("name")); - headerExpressions.put(headerName, expression); - } + return hasValue ? + new LiteralExpression(headerValue) : + EXPRESSION_PARSER.parseExpression(headerExpression); + })); gatewayMethodMetadata.setHeaderExpressions(headerExpressions); setGlobalMethodMetadata(gatewayMethodMetadata); } + } - if (StringUtils.hasText(mapper)) { - setMapper(beanFactory.getBean(mapper, MethodArgsMessageMapper.class)); - } - - String defaultRequestTimeout = - beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("defaultRequestTimeout")); - if (StringUtils.hasText(defaultRequestTimeout)) { - setDefaultRequestTimeout(Long.parseLong(defaultRequestTimeout)); - } - - String defaultReplyTimeout = - beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("defaultReplyTimeout")); - if (StringUtils.hasText(defaultReplyTimeout)) { - setDefaultReplyTimeout(Long.parseLong(defaultReplyTimeout)); - } - - super.onInit(); + @Nullable + private String resolveAttribute(String attributeName) { + ConfigurableListableBeanFactory beanFactory = (ConfigurableListableBeanFactory) getBeanFactory(); + return beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString(attributeName)); } }