From 761af2730cad8633eb28bfedecfa1670ce6d352c Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 19 Dec 2018 15:25:27 -0500 Subject: [PATCH] * Fix Sonar issues for Sec., STOMP, SFTP, WebFlux --- ...lSecurityInterceptorBeanPostProcessor.java | 24 ++- ...tyIntegrationConfigurationInitializer.java | 125 +++++++++------- .../sftp/gateway/SftpOutboundGateway.java | 38 ++--- .../stomp/AbstractStompSessionManager.java | 71 ++++----- .../inbound/StompInboundChannelAdapter.java | 80 +++++----- .../stomp/outbound/StompMessageHandler.java | 60 +++++--- .../stomp/support/StompHeaderMapper.java | 65 ++++---- .../inbound/WebFluxInboundEndpoint.java | 139 +++++++++--------- 8 files changed, 342 insertions(+), 260 deletions(-) diff --git a/spring-integration-security/src/main/java/org/springframework/integration/security/config/ChannelSecurityInterceptorBeanPostProcessor.java b/spring-integration-security/src/main/java/org/springframework/integration/security/config/ChannelSecurityInterceptorBeanPostProcessor.java index 270f944675..8814216572 100644 --- a/spring-integration-security/src/main/java/org/springframework/integration/security/config/ChannelSecurityInterceptorBeanPostProcessor.java +++ b/spring-integration-security/src/main/java/org/springframework/integration/security/config/ChannelSecurityInterceptorBeanPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -27,10 +27,12 @@ import org.springframework.aop.TargetSource; import org.springframework.aop.framework.autoproxy.AbstractAutoProxyCreator; import org.springframework.aop.support.DefaultBeanFactoryPointcutAdvisor; import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.integration.security.channel.ChannelAccessPolicy; import org.springframework.integration.security.channel.ChannelSecurityInterceptor; import org.springframework.integration.security.channel.ChannelSecurityMetadataSource; +import org.springframework.lang.Nullable; import org.springframework.messaging.MessageChannel; /** @@ -53,6 +55,7 @@ public class ChannelSecurityInterceptorBeanPostProcessor extends AbstractAutoPro public ChannelSecurityInterceptorBeanPostProcessor(Map> securityInterceptorMappings, Map> accessPolicyMapping) { + this.securityInterceptorMappings = securityInterceptorMappings; //NOSONAR (inconsistent sync) this.accessPolicyMapping = accessPolicyMapping; //NOSONAR (inconsistent sync) } @@ -73,19 +76,24 @@ public class ChannelSecurityInterceptorBeanPostProcessor extends AbstractAutoPro } @Override + @Nullable protected Object[] getAdvicesAndAdvisorsForBean(Class beanClass, String beanName, - TargetSource customTargetSource) throws BeansException { + @Nullable TargetSource customTargetSource) throws BeansException { + if (MessageChannel.class.isAssignableFrom(beanClass)) { - List interceptors = new ArrayList(); + List interceptors = new ArrayList<>(); for (Map.Entry> entry : this.securityInterceptorMappings.entrySet()) { if (isMatch(beanName, entry.getValue())) { - DefaultBeanFactoryPointcutAdvisor channelSecurityInterceptor - = new DefaultBeanFactoryPointcutAdvisor(); - channelSecurityInterceptor.setAdviceBeanName(entry.getKey()); - channelSecurityInterceptor.setBeanFactory(getBeanFactory()); - interceptors.add(channelSecurityInterceptor); + DefaultBeanFactoryPointcutAdvisor channelSecurityInterceptor = + new DefaultBeanFactoryPointcutAdvisor(); + channelSecurityInterceptor.setAdviceBeanName(entry.getKey()); + BeanFactory beanFactory = getBeanFactory(); + if (beanFactory != null) { + channelSecurityInterceptor.setBeanFactory(beanFactory); } + interceptors.add(channelSecurityInterceptor); } + } if (!interceptors.isEmpty()) { return interceptors.toArray(); } diff --git a/spring-integration-security/src/main/java/org/springframework/integration/security/config/SecurityIntegrationConfigurationInitializer.java b/spring-integration-security/src/main/java/org/springframework/integration/security/config/SecurityIntegrationConfigurationInitializer.java index e2b6860666..275b7482ed 100644 --- a/spring-integration-security/src/main/java/org/springframework/integration/security/config/SecurityIntegrationConfigurationInitializer.java +++ b/spring-integration-security/src/main/java/org/springframework/integration/security/config/SecurityIntegrationConfigurationInitializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2018 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. @@ -18,12 +18,14 @@ package org.springframework.integration.security.config; import java.util.HashMap; import java.util.Map; +import java.util.Set; import java.util.regex.Pattern; import org.springframework.beans.BeansException; import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.config.ConstructorArgumentValues; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.ManagedMap; @@ -39,6 +41,7 @@ import org.springframework.integration.security.channel.SecuredChannel; * The Integration Security infrastructure {@code beanFactory} initializer. * * @author Artem Bilan + * * @since 4.0 */ public class SecurityIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer { @@ -47,71 +50,30 @@ public class SecurityIntegrationConfigurationInitializer implements IntegrationC ChannelSecurityInterceptorBeanPostProcessor.class.getName(); @Override - @SuppressWarnings("unchecked") public void initialize(ConfigurableListableBeanFactory beanFactory) throws BeansException { BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; - Map> securityInterceptors = new ManagedMap>(); - Map> policies = new HashMap>(); + Map> securityInterceptors = new ManagedMap<>(); + Map> policies = new HashMap<>(); for (String beanName : registry.getBeanDefinitionNames()) { BeanDefinition beanDefinition = registry.getBeanDefinition(beanName); if (ChannelSecurityInterceptor.class.getName().equals(beanDefinition.getBeanClassName())) { - BeanDefinition metadataSource = (BeanDefinition) beanDefinition.getConstructorArgumentValues() - .getIndexedArgumentValue(0, BeanDefinition.class) - .getValue(); - - Map value = (Map) metadataSource.getConstructorArgumentValues() - .getIndexedArgumentValue(0, Map.class) - .getValue(); - ManagedSet patterns = new ManagedSet(); - if (!securityInterceptors.containsKey(beanName)) { - securityInterceptors.put(beanName, patterns); - } - else { - patterns = securityInterceptors.get(beanName); - } - patterns.addAll(value.keySet()); + collectPatternsFromInterceptor(securityInterceptors, beanName, beanDefinition); } else if (beanDefinition instanceof AnnotatedBeanDefinition) { - if (beanDefinition.getSource() instanceof MethodMetadata) { - MethodMetadata beanMethod = (MethodMetadata) beanDefinition.getSource(); - String annotationType = SecuredChannel.class.getName(); - if (beanMethod.isAnnotated(annotationType)) { - Map securedAttributes = beanMethod.getAnnotationAttributes(annotationType); - String[] interceptors = (String[]) securedAttributes.get("interceptor"); - String[] sendAccess = (String[]) securedAttributes.get("sendAccess"); - String[] receiveAccess = (String[]) securedAttributes.get("receiveAccess"); - ChannelAccessPolicy accessPolicy = new DefaultChannelAccessPolicy(sendAccess, receiveAccess); - for (String interceptor : interceptors) { - ManagedSet patterns = new ManagedSet(); - if (!securityInterceptors.containsKey(interceptor)) { - securityInterceptors.put(interceptor, patterns); - } - else { - patterns = securityInterceptors.get(interceptor); - } - patterns.add(beanName); - - Map mapping = new HashMap(); - if (!policies.containsKey(interceptor)) { - policies.put(interceptor, mapping); - } - else { - mapping = policies.get(interceptor); - } - mapping.put(Pattern.compile(beanName), accessPolicy); - } - } + Object beanSource = beanDefinition.getSource(); + if (beanSource instanceof MethodMetadata) { + collectInterceptorsAndPoliciesBySecuredChannel(securityInterceptors, policies, beanName, + (MethodMetadata) beanSource); } } } if (!securityInterceptors.isEmpty()) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(ChannelSecurityInterceptorBeanPostProcessor.class) - .addConstructorArgValue(securityInterceptors); + .addConstructorArgValue(securityInterceptors); if (!policies.isEmpty()) { builder.addConstructorArgValue(policies); } @@ -119,4 +81,67 @@ public class SecurityIntegrationConfigurationInitializer implements IntegrationC } } + @SuppressWarnings("unchecked") + private void collectPatternsFromInterceptor(Map> securityInterceptors, String beanName, + BeanDefinition beanDefinition) { + + ConstructorArgumentValues.ValueHolder metadataSourceValueHolder = + beanDefinition + .getConstructorArgumentValues() + .getIndexedArgumentValue(0, BeanDefinition.class); + if (metadataSourceValueHolder != null) { + BeanDefinition metadataSource = (BeanDefinition) metadataSourceValueHolder.getValue(); + if (metadataSource != null) { + ConstructorArgumentValues.ValueHolder patternMappingsValueHolder = + metadataSource + .getConstructorArgumentValues() + .getIndexedArgumentValue(0, Map.class); + if (patternMappingsValueHolder != null) { + Map patternsToAdd = (Map) patternMappingsValueHolder.getValue(); + Set patterns = new ManagedSet<>(); + if (!securityInterceptors.containsKey(beanName)) { + securityInterceptors.put(beanName, patterns); + } + else { + patterns = securityInterceptors.get(beanName); + } + if (patternsToAdd != null) { + patterns.addAll(patternsToAdd.keySet()); + } + } + } + } + } + + private void collectInterceptorsAndPoliciesBySecuredChannel(Map> securityInterceptors, + Map> policies, String beanName, MethodMetadata beanMethod) { + + Map securedAttributes = beanMethod.getAnnotationAttributes(SecuredChannel.class.getName()); + if (securedAttributes != null) { + String[] interceptors = (String[]) securedAttributes.get("interceptor"); + String[] sendAccess = (String[]) securedAttributes.get("sendAccess"); + String[] receiveAccess = (String[]) securedAttributes.get("receiveAccess"); + ChannelAccessPolicy accessPolicy = new DefaultChannelAccessPolicy(sendAccess, receiveAccess); + for (String interceptor : interceptors) { + Set patterns = new ManagedSet<>(); + if (!securityInterceptors.containsKey(interceptor)) { + securityInterceptors.put(interceptor, patterns); + } + else { + patterns = securityInterceptors.get(interceptor); + } + patterns.add(beanName); + + Map mapping = new HashMap<>(); + if (!policies.containsKey(interceptor)) { + policies.put(interceptor, mapping); + } + else { + mapping = policies.get(interceptor); + } + mapping.put(Pattern.compile(beanName), accessPolicy); + } + } + } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java index 7fff432cd4..bdbb012750 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -17,9 +17,9 @@ package org.springframework.integration.sftp.gateway; import java.lang.reflect.Method; -import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.stream.Collectors; import org.springframework.integration.file.remote.AbstractFileInfo; import org.springframework.integration.file.remote.ClientCallbackWithoutResult; @@ -42,6 +42,7 @@ import com.jcraft.jsch.SftpException; * * @author Gary Russell * @author Artem Bilan + * * @since 2.1 */ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway { @@ -49,8 +50,10 @@ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway sessionFactory, MessageSessionCallback messageSessionCallback) { + this(new SftpRemoteFileTemplate(sessionFactory), messageSessionCallback); } @@ -72,6 +76,7 @@ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway remoteFileTemplate, MessageSessionCallback messageSessionCallback) { + super(remoteFileTemplate, messageSessionCallback); } @@ -119,11 +124,9 @@ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway> asFileInfoList(Collection files) { - List> canonicalFiles = new ArrayList>(); - for (LsEntry file : files) { - canonicalFiles.add(new SftpFileInfo(file)); - } - return canonicalFiles; + return files.stream() + .map(SftpFileInfo::new) + .collect(Collectors.toList()); } @Override @@ -149,14 +152,15 @@ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway remoteFileOperations, final String path, final int chmod) { - remoteFileOperations.executeWithClient((ClientCallbackWithoutResult) client -> { - try { - client.chmod(chmod, path); - } - catch (SftpException e) { - throw new GeneralSftpException("Failed to execute chmod", e); - } - }); + remoteFileOperations + .executeWithClient((ClientCallbackWithoutResult) client -> { + try { + client.chmod(chmod, path); + } + catch (SftpException e) { + throw new GeneralSftpException("Failed to execute chmod", e); + } + }); } } diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java index 33c7a278f4..2a5205eb0f 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2017 the original author or authors. + * Copyright 2015-2018 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. @@ -35,12 +35,14 @@ import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.SmartLifecycle; import org.springframework.integration.stomp.event.StompConnectionFailedEvent; import org.springframework.integration.stomp.event.StompSessionConnectedEvent; +import org.springframework.lang.Nullable; import org.springframework.messaging.simp.stomp.StompClientSupport; import org.springframework.messaging.simp.stomp.StompCommand; import org.springframework.messaging.simp.stomp.StompHeaders; import org.springframework.messaging.simp.stomp.StompSession; import org.springframework.messaging.simp.stomp.StompSessionHandler; import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter; +import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.concurrent.ListenableFuture; @@ -73,12 +75,12 @@ public abstract class AbstractStompSessionManager implements StompSessionManager protected final Log logger = LogFactory.getLog(getClass()); + protected final StompClientSupport stompClient; + private final CompositeStompSessionHandler compositeStompSessionHandler = new CompositeStompSessionHandler(); private final Object lifecycleMonitor = new Object(); - protected final StompClientSupport stompClient; - private final AtomicInteger epoch = new AtomicInteger(); private boolean autoStartup = false; @@ -176,9 +178,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager private synchronized void connect() { if (this.connecting || this.connected) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Aborting connect; another thread is connecting."); - } + this.logger.debug("Aborting connect; another thread is connecting."); return; } final int epoch = this.epoch.get(); @@ -201,12 +201,12 @@ public abstract class AbstractStompSessionManager implements StompSessionManager final CountDownLatch connectLatch = new CountDownLatch(1); this.stompSessionListenableFuture.addCallback( stompSession -> { - if (AbstractStompSessionManager.this.logger.isDebugEnabled()) { - AbstractStompSessionManager.this.logger.debug("onSuccess"); - } + AbstractStompSessionManager.this.logger.debug("onSuccess"); AbstractStompSessionManager.this.connected = true; AbstractStompSessionManager.this.connecting = false; - stompSession.setAutoReceipt(isAutoReceiptEnabled()); + if (stompSession != null) { + stompSession.setAutoReceipt(isAutoReceiptEnabled()); + } if (AbstractStompSessionManager.this.applicationEventPublisher != null) { AbstractStompSessionManager.this.applicationEventPublisher.publishEvent( new StompSessionConnectedEvent(this)); @@ -216,9 +216,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager }, e -> { - if (AbstractStompSessionManager.this.logger.isDebugEnabled()) { - AbstractStompSessionManager.this.logger.debug("onFailure", e); - } + AbstractStompSessionManager.this.logger.debug("onFailure", e); connectLatch.countDown(); if (epoch == AbstractStompSessionManager.this.epoch.get()) { scheduleReconnect(e); @@ -255,12 +253,14 @@ public abstract class AbstractStompSessionManager implements StompSessionManager this.reconnectFuture = null; } - if (this.stompClient.getTaskScheduler() != null) { - this.reconnectFuture = this.stompClient.getTaskScheduler() - .schedule(this::connect, new Date(System.currentTimeMillis() + this.recoveryInterval)); + TaskScheduler taskScheduler = this.stompClient.getTaskScheduler(); + if (taskScheduler != null) { + this.reconnectFuture = + taskScheduler.schedule(this::connect, + new Date(System.currentTimeMillis() + this.recoveryInterval)); } else { - this.logger.info("For automatic reconnection the 'stompClient' should be configured with a TaskScheduler."); + this.logger.info("For automatic reconnection the stompClient should be configured with a TaskScheduler."); } } @@ -271,20 +271,23 @@ public abstract class AbstractStompSessionManager implements StompSessionManager this.reconnectFuture.cancel(false); this.reconnectFuture = null; } - this.stompSessionListenableFuture.addCallback(new ListenableFutureCallback() { + this.stompSessionListenableFuture.addCallback( + new ListenableFutureCallback() { - @Override - public void onFailure(Throwable ex) { - AbstractStompSessionManager.this.connected = false; - } + @Override + public void onFailure(Throwable ex) { + AbstractStompSessionManager.this.connected = false; + } - @Override - public void onSuccess(StompSession session) { - session.disconnect(); - AbstractStompSessionManager.this.connected = false; - } + @Override + public void onSuccess(StompSession session) { + if (session != null) { + session.disconnect(); + } + AbstractStompSessionManager.this.connected = false; + } - }); + }); this.stompSessionListenableFuture = null; } } @@ -294,7 +297,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager synchronized (this.lifecycleMonitor) { if (!isRunning()) { if (this.logger.isInfoEnabled()) { - this.logger.info("Starting " + getClass().getSimpleName()); + this.logger.info("Starting " + this); } connect(); this.running = true; @@ -318,7 +321,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager if (isRunning()) { this.running = false; if (this.logger.isInfoEnabled()) { - this.logger.info("Stopping " + getClass().getSimpleName()); + this.logger.info("Stopping " + this); } destroy(); } @@ -360,8 +363,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager private class CompositeStompSessionHandler extends StompSessionHandlerAdapter { - private final List delegates = - Collections.synchronizedList(new ArrayList()); + private final List delegates = Collections.synchronizedList(new ArrayList<>()); private volatile StompSession session; @@ -393,8 +395,9 @@ public abstract class AbstractStompSessionManager implements StompSessionManager } @Override - public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, - Throwable exception) { + public void handleException(StompSession session, @Nullable StompCommand command, StompHeaders headers, + byte[] payload, Throwable exception) { + synchronized (this.delegates) { for (StompSessionHandler delegate : this.delegates) { delegate.handleException(session, command, headers, payload, exception); diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java index 7a182465ec..a14a5b33ad 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2018 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. @@ -36,6 +36,7 @@ import org.springframework.integration.support.management.IntegrationManagedReso import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; @@ -60,6 +61,7 @@ import org.springframework.util.Assert; * if provided {@link StompSessionManager} supports {@code autoReceiptEnabled}. * * @author Artem Bilan + * * @since 4.2 */ @ManagedResource @@ -68,25 +70,22 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement private final StompSessionHandler stompSessionHandler = new IntegrationInboundStompSessionHandler(); - private final Set destinations = new LinkedHashSet(); + private final Set destinations = new LinkedHashSet<>(); private final StompSessionManager stompSessionManager; - private final Map subscriptions = - new HashMap(); + private final Map subscriptions = new HashMap<>(); private final Lock destinationLock = new ReentrantLock(); private ApplicationEventPublisher applicationEventPublisher; + private Class payloadType = String.class; + + private HeaderMapper headerMapper = new StompHeaderMapper(); + private volatile StompSession stompSession; - private volatile Class payloadType = String.class; - - private volatile HeaderMapper headerMapper = new StompHeaderMapper(); - - private volatile MessageChannel errorChannel; - public StompInboundChannelAdapter(StompSessionManager stompSessionManager, String... destinations) { Assert.notNull(stompSessionManager, "'stompSessionManager' is required."); if (destinations != null) { @@ -103,12 +102,6 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement this.payloadType = payloadType; } - @Override - public void setErrorChannel(MessageChannel errorChannel) { - super.setErrorChannel(errorChannel); - this.errorChannel = errorChannel; - } - public void setHeaderMapper(HeaderMapper headerMapper) { Assert.notNull(headerMapper, "'headerMapper' must not be null."); this.headerMapper = headerMapper; @@ -123,7 +116,7 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement public String[] getDestinations() { this.destinationLock.lock(); try { - return this.destinations.toArray(new String[this.destinations.size()]); + return this.destinations.toArray(new String[0]); } finally { this.destinationLock.unlock(); @@ -206,7 +199,7 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement } } catch (Exception e) { - logger.warn("The exception during unsubscription.", e); + logger.warn("The exception during unsubscribing.", e); } this.subscriptions.clear(); } @@ -222,15 +215,24 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement } @Override - public void handleFrame(StompHeaders headers, Object body) { + public void handleFrame(StompHeaders headers, @Nullable Object body) { Message message; - if (body instanceof Message) { + + if (body == null) { + logger.info("No body in STOMP frame: nothing to produce."); + return; + } + else if (body instanceof Message) { message = (Message) body; } else { - message = getMessageBuilderFactory().withPayload(body) - .copyHeaders(StompInboundChannelAdapter.this.headerMapper.toHeaders(headers)) - .build(); + Map headersToCopy = + StompInboundChannelAdapter.this.headerMapper.toHeaders(headers); + message = + getMessageBuilderFactory() + .withPayload(body) + .copyHeaders(headersToCopy) + .build(); } sendMessage(message); } @@ -260,7 +262,7 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement } this.subscriptions.put(destination, subscription); } - else { + else if (logger.isWarnEnabled()) { logger.warn("The StompInboundChannelAdapter [" + getComponentName() + "] ins't connected to StompSession. Check the state of [" + this.stompSessionManager + "]"); } @@ -277,15 +279,27 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement } @Override - public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, - Throwable exception) { - if (StompInboundChannelAdapter.this.errorChannel != null) { - StompHeaderAccessor headerAccessor = StompHeaderAccessor.create(command); - headerAccessor.copyHeaders(StompInboundChannelAdapter.this.headerMapper.toHeaders(headers)); - Message failedMessage = MessageBuilder.createMessage(payload, - headerAccessor.getMessageHeaders()); - getMessagingTemplate().send(StompInboundChannelAdapter.this.errorChannel, - new ErrorMessage(new MessageHandlingException(failedMessage, exception))); + public void handleException(StompSession session, @Nullable StompCommand command, StompHeaders headers, + byte[] payload, Throwable exception) { + + MessageChannel errorChannel = getErrorChannel(); + if (errorChannel != null) { + Message failedMessage; + // TODO 5.2 Copy all the STOMP headers for error message without any mapping + Map headersToCopy = StompInboundChannelAdapter.this.headerMapper.toHeaders(headers); + if (command != null) { + StompHeaderAccessor headerAccessor = StompHeaderAccessor.create(command); + headerAccessor.copyHeaders(headersToCopy); + failedMessage = MessageBuilder.createMessage(payload, headerAccessor.getMessageHeaders()); + } + else { + failedMessage = + MessageBuilder.withPayload(payload) + .copyHeaders(headersToCopy) + .build(); + } + getMessagingTemplate() + .send(errorChannel, new ErrorMessage(new MessageHandlingException(failedMessage, exception))); } else { logger.error("STOMP Frame handling error.", exception); diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java index 9bea8b6501..c4231381af 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2018 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. @@ -32,6 +32,7 @@ import org.springframework.integration.stomp.StompSessionManager; import org.springframework.integration.stomp.event.StompExceptionEvent; import org.springframework.integration.stomp.event.StompReceiptEvent; import org.springframework.integration.stomp.support.StompHeaderMapper; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessagingException; @@ -47,7 +48,9 @@ import org.springframework.util.Assert; /** * The {@link AbstractMessageHandler} implementation to send messages to STOMP destinations. + * * @author Artem Bilan + * * @since 4.2 */ public class StompMessageHandler extends AbstractMessageHandler implements ApplicationEventPublisherAware, Lifecycle { @@ -60,13 +63,7 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli private final Semaphore connectSemaphore = new Semaphore(0); - private volatile StompSession stompSession; - - private volatile Throwable transportError; - - private volatile boolean running; - - private volatile HeaderMapper headerMapper = new StompHeaderMapper(); + private HeaderMapper headerMapper = new StompHeaderMapper(); private Expression destinationExpression; @@ -74,7 +71,13 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli private ApplicationEventPublisher applicationEventPublisher; - private volatile long connectTimeout = DEFAULT_CONNECT_TIMEOUT; + private long connectTimeout = DEFAULT_CONNECT_TIMEOUT; + + private volatile StompSession stompSession; + + private volatile Throwable transportError; + + private volatile boolean running; public StompMessageHandler(StompSessionManager stompSessionManager) { Assert.notNull(stompSessionManager, "'stompSessionManager' is required."); @@ -126,19 +129,20 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli } @Override - protected void handleMessageInternal(final Message message) throws Exception { + protected void handleMessageInternal(final Message message) { try { connectIfNecessary(); } catch (Exception e) { - throw new MessageDeliveryException(message, "The [" + this + "] could not deliver message.", e); + throw new MessageDeliveryException(message, "The '" + this + "' could not deliver message.", e); } StompSession stompSession = this.stompSession; StompHeaders stompHeaders = new StompHeaders(); this.headerMapper.fromHeaders(message.getHeaders(), stompHeaders); if (stompHeaders.getDestination() == null) { - Assert.state(this.destinationExpression != null, "One of 'destination' or 'destinationExpression' must be" + + Assert.state(this.destinationExpression != null, "One of 'destination' or 'destinationExpression' must " + + "be" + " provided, if message header doesn't supply 'destination' STOMP header."); String destination = this.destinationExpression.getValue(this.evaluationContext, message, String.class); stompHeaders.setDestination(destination); @@ -171,7 +175,7 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli } } - private StompSession connectIfNecessary() throws Exception { + private void connectIfNecessary() throws InterruptedException { synchronized (this.connectSemaphore) { if (this.stompSession == null || !this.stompSessionManager.isConnected()) { this.stompSessionManager.disconnect(this.sessionHandler); @@ -192,7 +196,6 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli } } } - return this.stompSession; } } @@ -235,22 +238,35 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli Message failedMessage = getMessageBuilderFactory().withPayload(thePayload) .copyHeaders(StompMessageHandler.this.headerMapper.toHeaders(headers)) .build(); - MessagingException exception = new MessageDeliveryException(failedMessage, - "STOMP frame handling error."); - logger.error("STOMP frame handling error.", exception); + MessagingException exception = + new MessageDeliveryException(failedMessage, "STOMP frame handling error."); + if (StompMessageHandler.this.applicationEventPublisher != null) { StompMessageHandler.this.applicationEventPublisher.publishEvent( new StompExceptionEvent(StompMessageHandler.this, exception)); } + else { + logger.error(exception); + } } } @Override - public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, - Throwable exception) { - Message message = MessageBuilder.createMessage(payload, - StompHeaderAccessor.create(command, headers).getMessageHeaders()); - logger.error("The exception for session [" + session + "] on message [" + message + "]", exception); + public void handleException(StompSession session, @Nullable StompCommand command, + StompHeaders headers, byte[] payload, Throwable exception) { + + Message failedMessage; + if (command != null) { + StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.create(command, headers); + failedMessage = MessageBuilder.createMessage(payload, stompHeaderAccessor.getMessageHeaders()); + } + else { + failedMessage = + MessageBuilder.withPayload(payload) + .copyHeaders(headers) + .build(); + } + logger.error("The exception for session [" + session + "] on message [" + failedMessage + "]", exception); } @Override diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/StompHeaderMapper.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/StompHeaderMapper.java index 97a719a61d..04e90ade23 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/StompHeaderMapper.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/StompHeaderMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2018 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. @@ -40,7 +40,9 @@ import org.springframework.util.StringUtils; * The STOMP {@link HeaderMapper} implementation. * * @author Artem Bilan + * * @since 4.2 + * * @see StompHeaders */ public class StompHeaderMapper implements HeaderMapper { @@ -51,32 +53,32 @@ public class StompHeaderMapper implements HeaderMapper { public static final String STOMP_OUTBOUND_HEADER_NAME_PATTERN = "STOMP_OUTBOUND_HEADERS"; - private static final String[] STOMP_INBOUND_HEADER_NAMES = new String[] { - StompHeaders.CONTENT_LENGTH, - StompHeaders.CONTENT_TYPE, - StompHeaders.MESSAGE_ID, - StompHeaders.RECEIPT_ID, - StompHeaders.SUBSCRIPTION, - }; + private static final String[] STOMP_INBOUND_HEADER_NAMES = + new String[] { + StompHeaders.CONTENT_LENGTH, + StompHeaders.CONTENT_TYPE, + StompHeaders.MESSAGE_ID, + StompHeaders.RECEIPT_ID, + StompHeaders.SUBSCRIPTION, + }; - private final static List STOMP_INBOUND_HEADER_NAMES_LIST = - Arrays.asList(STOMP_INBOUND_HEADER_NAMES); + private static final List STOMP_INBOUND_HEADER_NAMES_LIST = Arrays.asList(STOMP_INBOUND_HEADER_NAMES); - private static final String[] STOMP_OUTBOUND_HEADER_NAMES = new String[] { - StompHeaders.CONTENT_LENGTH, - StompHeaders.CONTENT_TYPE, - StompHeaders.DESTINATION, - StompHeaders.RECEIPT, - IntegrationStompHeaders.DESTINATION, - IntegrationStompHeaders.RECEIPT - }; + private static final String[] STOMP_OUTBOUND_HEADER_NAMES = + new String[] { + StompHeaders.CONTENT_LENGTH, + StompHeaders.CONTENT_TYPE, + StompHeaders.DESTINATION, + StompHeaders.RECEIPT, + IntegrationStompHeaders.DESTINATION, + IntegrationStompHeaders.RECEIPT + }; - private final static List STOMP_OUTBOUND_HEADER_NAMES_LIST = - Arrays.asList(STOMP_OUTBOUND_HEADER_NAMES); + private static final List STOMP_OUTBOUND_HEADER_NAMES_LIST = Arrays.asList(STOMP_OUTBOUND_HEADER_NAMES); - private volatile String[] inboundHeaderNames = STOMP_INBOUND_HEADER_NAMES; + private String[] inboundHeaderNames = STOMP_INBOUND_HEADER_NAMES; - private volatile String[] outboundHeaderNames = STOMP_OUTBOUND_HEADER_NAMES; + private String[] outboundHeaderNames = STOMP_OUTBOUND_HEADER_NAMES; public void setInboundHeaderNames(String[] inboundHeaderNames) { //NOSONAR - false positive Assert.notNull(inboundHeaderNames, "'inboundHeaderNames' must not be null."); @@ -110,12 +112,14 @@ public class StompHeaderMapper implements HeaderMapper { else if (StompHeaderAccessor.NATIVE_HEADERS.equals(name)) { MultiValueMap multiValueMap = headers.get(StompHeaderAccessor.NATIVE_HEADERS, MultiValueMap.class); - for (Map.Entry> entry1 : multiValueMap.entrySet()) { - name = entry1.getKey(); - if (shouldMapHeader(name, this.outboundHeaderNames)) { - String value = entry1.getValue().get(0); - if (StringUtils.hasText(value)) { - setStompHeader(target, name, value); + if (multiValueMap != null) { + for (Map.Entry> entry1 : multiValueMap.entrySet()) { + name = entry1.getKey(); + if (shouldMapHeader(name, this.outboundHeaderNames)) { + String value = entry1.getValue().get(0); + if (StringUtils.hasText(value)) { + setStompHeader(target, name, value); + } } } } @@ -149,7 +153,8 @@ public class StompHeaderMapper implements HeaderMapper { else { Class clazz = (value != null) ? value.getClass() : null; throw new IllegalArgumentException( - "Expected MediaType or String value for 'content-type' header value, but received: " + clazz); + "Expected MediaType or String value for 'content-type' header value, but received: " + + clazz); } } } @@ -187,7 +192,7 @@ public class StompHeaderMapper implements HeaderMapper { @Override public Map toHeaders(StompHeaders source) { - Map target = new HashMap(); + Map target = new HashMap<>(); for (String name : source.keySet()) { if (shouldMapHeader(name, this.inboundHeaderNames)) { if (StompHeaders.CONTENT_TYPE.equals(name)) { diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java index fa7991a74e..ca847f7d32 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java @@ -33,6 +33,7 @@ import org.reactivestreams.Publisher; import org.springframework.core.ReactiveAdapter; import org.springframework.core.ReactiveAdapterRegistry; import org.springframework.core.ResolvableType; +import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.http.HttpHeaders; @@ -147,6 +148,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W private Mono doHandle(ServerWebExchange exchange) { return extractRequestBody(exchange) .doOnSubscribe(s -> this.activeCount.incrementAndGet()) + .cast(Object.class) .switchIfEmpty(Mono.just(exchange.getRequest().getQueryParams())) .map(body -> new RequestEntity<>(body, exchange.getRequest().getHeaders(), @@ -166,71 +168,77 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W } - @SuppressWarnings(UNCHECKED) - private Mono extractRequestBody(ServerWebExchange exchange) { - ServerHttpRequest request = exchange.getRequest(); - ServerHttpResponse response = exchange.getResponse(); - - if (isReadable(request)) { - MediaType contentType; - if (request.getHeaders().getContentType() == null) { - contentType = MediaType.APPLICATION_OCTET_STREAM; - } - else { - contentType = request.getHeaders().getContentType(); - } - - if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(contentType)) { - return (Mono) exchange.getFormData(); - } - else if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) { - return (Mono) exchange.getMultipartData(); - } - else { - ResolvableType bodyType = getRequestPayloadType(); - if (bodyType == null) { - bodyType = - "text".equals(contentType.getType()) - ? ResolvableType.forClass(String.class) - : ResolvableType.forClass(byte[].class); - } - - Class resolvedType = bodyType.resolve(); - - ReactiveAdapter adapter = (resolvedType != null ? this.adapterRegistry.getAdapter(resolvedType) : - null); - ResolvableType elementType = (adapter != null ? bodyType.getGeneric() : bodyType); - - HttpMessageReader httpMessageReader = this.codecConfigurer - .getReaders() - .stream() - .filter(reader -> reader.canRead(elementType, contentType)) - .findFirst() - .orElseThrow(() -> new UnsupportedMediaTypeStatusException( - "Could not convert request: no suitable HttpMessageReader found for expected type [" - + elementType + "] and content type [" + contentType + "]")); - - - Map readHints = Collections.emptyMap(); - if (adapter != null && adapter.isMultiValue()) { - Flux flux = httpMessageReader.read(bodyType, elementType, request, response, readHints); - - return (Mono) Mono.just(adapter.fromPublisher(flux)); - } - else { - Mono mono = httpMessageReader.readMono(bodyType, elementType, request, response, readHints); - - if (adapter != null) { - return (Mono) Mono.just(adapter.fromPublisher(mono)); - } - else { - return (Mono) mono; - } - } - } + private Mono extractRequestBody(ServerWebExchange exchange) { + if (isReadable(exchange.getRequest())) { + return extractReadableRequestBody(exchange); } else { - return (Mono) Mono.just(exchange.getRequest().getQueryParams()); + return Mono.just(exchange.getRequest().getQueryParams()); + } + } + + private Mono extractReadableRequestBody(ServerWebExchange exchange) { + MediaType contentType = + exchange.getRequest() + .getHeaders() + .getContentType(); + if (contentType == null) { + contentType = MediaType.APPLICATION_OCTET_STREAM; + } + + if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(contentType)) { + return exchange.getFormData(); + } + else if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) { + return exchange.getMultipartData(); + } + else { + return readRequestBody(exchange, contentType); + } + } + + private Mono readRequestBody(ServerWebExchange exchange, MediaType contentType) { + ServerHttpRequest request = exchange.getRequest(); + ServerHttpResponse response = exchange.getResponse(); + ResolvableType bodyType = getRequestPayloadType(); + if (bodyType == null) { + bodyType = + "text".equals(contentType.getType()) + ? ResolvableType.forClass(String.class) + : ResolvableType.forClass(byte[].class); + } + + Class resolvedType = bodyType.resolve(); + + ReactiveAdapter adapter = + resolvedType != null + ? this.adapterRegistry.getAdapter(resolvedType) + : null; + ResolvableType elementType = (adapter != null ? bodyType.getGeneric() : bodyType); + + HttpMessageReader httpMessageReader = this.codecConfigurer + .getReaders() + .stream() + .filter(reader -> reader.canRead(elementType, contentType)) + .findFirst() + .orElseThrow(() -> new UnsupportedMediaTypeStatusException( + "Could not convert request: no suitable HttpMessageReader found for expected type [" + + elementType + "] and content type [" + contentType + "]")); + + + Map readHints = Collections.emptyMap(); + if (adapter != null && adapter.isMultiValue()) { + Flux flux = httpMessageReader.read(bodyType, elementType, request, response, readHints); + return Mono.just(adapter.fromPublisher(flux)); + } + else { + Mono mono = httpMessageReader.readMono(bodyType, elementType, request, response, readHints); + if (adapter != null) { + return Mono.just(adapter.fromPublisher(mono)); + } + else { + return mono; + } } } @@ -241,7 +249,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W ServerHttpRequest request = exchange.getRequest(); MultiValueMap requestParams = request.getQueryParams(); - StandardEvaluationContext evaluationContext = buildEvaluationContext(httpEntity, exchange); + EvaluationContext evaluationContext = buildEvaluationContext(httpEntity, exchange); Object payload; if (getPayloadExpression() != null) { payload = getPayloadExpression().getValue(evaluationContext); @@ -299,7 +307,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W } @SuppressWarnings(UNCHECKED) - private StandardEvaluationContext buildEvaluationContext(RequestEntity httpEntity, ServerWebExchange exchange) { + private EvaluationContext buildEvaluationContext(RequestEntity httpEntity, ServerWebExchange exchange) { ServerHttpRequest request = exchange.getRequest(); HttpHeaders requestHeaders = request.getHeaders(); MultiValueMap requestParams = request.getQueryParams(); @@ -482,7 +490,6 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W return (mediaTypes.isEmpty() ? Collections.singletonList(MediaType.ALL) : mediaTypes); } - @SuppressWarnings(UNCHECKED) private List getProducibleTypes(ServerWebExchange exchange, Supplier> producibleTypesSupplier) {