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

* Fix Sonar issues for Sec., STOMP, SFTP, WebFlux

Browse Source
This commit is contained in:
Artem Bilan
2018-12-19 15:25:27 -05:00
committed by Gary Russell
parent 7790f9e550
commit 761af2730c
8 changed files with 342 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-integration-security/src/main/java/org/springframework/integration/security/config/ChannelSecurityInterceptorBeanPostProcessor.java
View File

@@ -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<String, Set<Pattern>> securityInterceptorMappings,
Map<String, Map<Pattern, ChannelAccessPolicy>> 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<Advisor> interceptors = new ArrayList<Advisor>();
List<Advisor> interceptors = new ArrayList<>();
for (Map.Entry<String, Set<Pattern>> 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();
}

125
spring-integration-security/src/main/java/org/springframework/integration/security/config/SecurityIntegrationConfigurationInitializer.java
View File

@@ -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<String, ManagedSet<String>> securityInterceptors = new ManagedMap<String, ManagedSet<String>>();
Map<String, Map<Pattern, ChannelAccessPolicy>> policies = new HashMap<String, Map<Pattern, ChannelAccessPolicy>>();
Map<String, Set<String>> securityInterceptors = new ManagedMap<>();
Map<String, Map<Pattern, ChannelAccessPolicy>> 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<String, ?> value = (Map<String, ?>) metadataSource.getConstructorArgumentValues()
.getIndexedArgumentValue(0, Map.class)
.getValue();
ManagedSet<String> patterns = new ManagedSet<String>();
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<String, Object> 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<String> patterns = new ManagedSet<String>();
if (!securityInterceptors.containsKey(interceptor)) {
securityInterceptors.put(interceptor, patterns);
}
else {
patterns = securityInterceptors.get(interceptor);
}
patterns.add(beanName);
Map<Pattern, ChannelAccessPolicy> mapping = new HashMap<Pattern, ChannelAccessPolicy>();
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<String, Set<String>> 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<String, ?> patternsToAdd = (Map<String, ?>) patternMappingsValueHolder.getValue();
Set<String> 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<String, Set<String>> securityInterceptors,
Map<String, Map<Pattern, ChannelAccessPolicy>> policies, String beanName, MethodMetadata beanMethod) {
Map<String, Object> 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<String> patterns = new ManagedSet<>();
if (!securityInterceptors.containsKey(interceptor)) {
securityInterceptors.put(interceptor, patterns);
}
else {
patterns = securityInterceptors.get(interceptor);
}
patterns.add(beanName);
Map<Pattern, ChannelAccessPolicy> mapping = new HashMap<>();
if (!policies.containsKey(interceptor)) {
policies.put(interceptor, mapping);
}
else {
mapping = policies.get(interceptor);
}
mapping.put(Pattern.compile(beanName), accessPolicy);
}
}
}
}

38
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java
View File

@@ -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<LsEntry> {
@@ -49,8 +50,10 @@ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway<LsEnt
private static final Method LS_ENTRY_SET_FILENAME_METHOD;
static {
LS_ENTRY_SET_FILENAME_METHOD = ReflectionUtils.findMethod(LsEntry.class, "setFilename", String.class);
LS_ENTRY_SET_FILENAME_METHOD.setAccessible(true);
LS_ENTRY_SET_FILENAME_METHOD = ReflectionUtils.findMethod(LsEntry.class, "setFilename", String.class);
if (LS_ENTRY_SET_FILENAME_METHOD != null) {
ReflectionUtils.makeAccessible(LS_ENTRY_SET_FILENAME_METHOD);
}
}
/**
@@ -61,6 +64,7 @@ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway<LsEnt
*/
public SftpOutboundGateway(SessionFactory<LsEntry> sessionFactory,
MessageSessionCallback<LsEntry, ?> messageSessionCallback) {
this(new SftpRemoteFileTemplate(sessionFactory), messageSessionCallback);
}
@@ -72,6 +76,7 @@ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway<LsEnt
*/
public SftpOutboundGateway(RemoteFileTemplate<LsEntry> remoteFileTemplate,
MessageSessionCallback<LsEntry, ?> messageSessionCallback) {
super(remoteFileTemplate, messageSessionCallback);
}
@@ -119,11 +124,9 @@ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway<LsEnt
@Override
protected List<AbstractFileInfo<LsEntry>> asFileInfoList(Collection<LsEntry> files) {
List<AbstractFileInfo<LsEntry>> canonicalFiles = new ArrayList<AbstractFileInfo<LsEntry>>();
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<LsEnt
@Override
protected void doChmod(RemoteFileOperations<LsEntry> remoteFileOperations, final String path, final int chmod) {
remoteFileOperations.executeWithClient((ClientCallbackWithoutResult<ChannelSftp>) client -> {
try {
client.chmod(chmod, path);
}
catch (SftpException e) {
throw new GeneralSftpException("Failed to execute chmod", e);
}
});
remoteFileOperations
.executeWithClient((ClientCallbackWithoutResult<ChannelSftp>) client -> {
try {
client.chmod(chmod, path);
}
catch (SftpException e) {
throw new GeneralSftpException("Failed to execute chmod", e);
}
});
}
}

71
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java
View File

@@ -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<StompSession>() {
this.stompSessionListenableFuture.addCallback(
new ListenableFutureCallback<StompSession>() {
@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<StompSessionHandler> delegates =
Collections.synchronizedList(new ArrayList<StompSessionHandler>());
private final List<StompSessionHandler> 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);

80
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java
View File

@@ -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<String> destinations = new LinkedHashSet<String>();
private final Set<String> destinations = new LinkedHashSet<>();
private final StompSessionManager stompSessionManager;
private final Map<String, StompSession.Subscription> subscriptions =
new HashMap<String, StompSession.Subscription>();
private final Map<String, StompSession.Subscription> subscriptions = new HashMap<>();
private final Lock destinationLock = new ReentrantLock();
private ApplicationEventPublisher applicationEventPublisher;
private Class<?> payloadType = String.class;
private HeaderMapper<StompHeaders> headerMapper = new StompHeaderMapper();
private volatile StompSession stompSession;
private volatile Class<?> payloadType = String.class;
private volatile HeaderMapper<StompHeaders> 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<StompHeaders> 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<String, Object> 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<byte[]> 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<byte[]> failedMessage;
// TODO 5.2 Copy all the STOMP headers for error message without any mapping
Map<String, Object> 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);

60
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java
View File

@@ -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<StompHeaders> headerMapper = new StompHeaderMapper();
private HeaderMapper<StompHeaders> 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<byte[]> 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<byte[]> 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

65
spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/StompHeaderMapper.java
View File

@@ -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<StompHeaders> {
@@ -51,32 +53,32 @@ public class StompHeaderMapper implements HeaderMapper<StompHeaders> {
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<String> STOMP_INBOUND_HEADER_NAMES_LIST =
Arrays.<String>asList(STOMP_INBOUND_HEADER_NAMES);
private static final List<String> 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<String> STOMP_OUTBOUND_HEADER_NAMES_LIST =
Arrays.<String>asList(STOMP_OUTBOUND_HEADER_NAMES);
private static final List<String> 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<StompHeaders> {
else if (StompHeaderAccessor.NATIVE_HEADERS.equals(name)) {
MultiValueMap<String, String> multiValueMap =
headers.get(StompHeaderAccessor.NATIVE_HEADERS, MultiValueMap.class);
for (Map.Entry<String, List<String>> 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<String, List<String>> 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<StompHeaders> {
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<StompHeaders> {
@Override
public Map<String, Object> toHeaders(StompHeaders source) {
Map<String, Object> target = new HashMap<String, Object>();
Map<String, Object> target = new HashMap<>();
for (String name : source.keySet()) {
if (shouldMapHeader(name, this.inboundHeaderNames)) {
if (StompHeaders.CONTENT_TYPE.equals(name)) {

139
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java
View File

@@ -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<Void> 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 <T> Mono<T> 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<T>) exchange.getFormData();
}
else if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) {
return (Mono<T>) 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<String, Object> readHints = Collections.emptyMap();
if (adapter != null && adapter.isMultiValue()) {
Flux<?> flux = httpMessageReader.read(bodyType, elementType, request, response, readHints);
return (Mono<T>) Mono.just(adapter.fromPublisher(flux));
}
else {
Mono<?> mono = httpMessageReader.readMono(bodyType, elementType, request, response, readHints);
if (adapter != null) {
return (Mono<T>) Mono.just(adapter.fromPublisher(mono));
}
else {
return (Mono<T>) mono;
}
}
}
private Mono<?> extractRequestBody(ServerWebExchange exchange) {
if (isReadable(exchange.getRequest())) {
return extractReadableRequestBody(exchange);
}
else {
return (Mono<T>) 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<String, Object> 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<String, String> 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<String, String> 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<MediaType> getProducibleTypes(ServerWebExchange exchange,
Supplier<List<MediaType>> producibleTypesSupplier) {