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

@Nullable all the way: null-safety at field level

Browse Source 
This commits extends nullability declarations to the field level, formalizing the interaction between methods and their underlying fields and therefore avoiding any nullability mismatch.

Issue: SPR-15720
This commit is contained in:
Juergen Hoeller
2017-06-30 01:53:45 +02:00
parent c4694c3f5c
commit cc74a2891a
936 changed files with 6090 additions and 2806 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-messaging/src/main/java/org/springframework/messaging/MessageHeaders.java
View File

@@ -96,7 +96,8 @@ public class MessageHeaders implements Map<String, Object>, Serializable {
private static final IdGenerator defaultIdGenerator = new AlternativeJdkIdGenerator();
private static volatile IdGenerator idGenerator = null;
@Nullable
private static volatile IdGenerator idGenerator;
private final Map<String, Object> headers;
@@ -162,7 +163,8 @@ public class MessageHeaders implements Map<String, Object>, Serializable {
}
protected static IdGenerator getIdGenerator() {
return (idGenerator != null ? idGenerator : defaultIdGenerator);
IdGenerator generator = idGenerator;
return (generator != null ? generator : defaultIdGenerator);
}
@Nullable

2
spring-messaging/src/main/java/org/springframework/messaging/MessagingException.java
View File

@@ -29,6 +29,7 @@ import org.springframework.lang.Nullable;
@SuppressWarnings("serial")
public class MessagingException extends NestedRuntimeException {
@Nullable
private final Message<?> failedMessage;
@@ -63,6 +64,7 @@ public class MessagingException extends NestedRuntimeException {
}
@Nullable
public Message<?> getFailedMessage() {
return this.failedMessage;
}

4
spring-messaging/src/main/java/org/springframework/messaging/converter/AbstractMessageConverter.java
View File

@@ -49,6 +49,7 @@ public abstract class AbstractMessageConverter implements SmartMessageConverter
private final List<MimeType> supportedMimeTypes;
@Nullable
private ContentTypeResolver contentTypeResolver = new DefaultContentTypeResolver();
private boolean strictContentTypeMatch = false;
@@ -91,13 +92,14 @@ public abstract class AbstractMessageConverter implements SmartMessageConverter
* ignore all messages.
* <p>By default, a {@code DefaultContentTypeResolver} instance is used.
*/
public void setContentTypeResolver(ContentTypeResolver resolver) {
public void setContentTypeResolver(@Nullable ContentTypeResolver resolver) {
this.contentTypeResolver = resolver;
}
/**
* Return the configured {@link ContentTypeResolver}.
*/
@Nullable
public ContentTypeResolver getContentTypeResolver() {
return this.contentTypeResolver;
}

14
spring-messaging/src/main/java/org/springframework/messaging/converter/MappingJackson2MessageConverter.java
View File

@@ -64,6 +64,7 @@ public class MappingJackson2MessageConverter extends AbstractMessageConverter {
private ObjectMapper objectMapper;
@Nullable
private Boolean prettyPrint;
@@ -73,7 +74,7 @@ public class MappingJackson2MessageConverter extends AbstractMessageConverter {
*/
public MappingJackson2MessageConverter() {
super(new MimeType("application", "json", StandardCharsets.UTF_8));
initObjectMapper();
this.objectMapper = initObjectMapper();
}
/**
@@ -84,14 +85,15 @@ public class MappingJackson2MessageConverter extends AbstractMessageConverter {
*/
public MappingJackson2MessageConverter(MimeType... supportedMimeTypes) {
super(Arrays.asList(supportedMimeTypes));
initObjectMapper();
this.objectMapper = initObjectMapper();
}
private void initObjectMapper() {
this.objectMapper = new ObjectMapper();
this.objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
this.objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
private ObjectMapper initObjectMapper() {
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
return objectMapper;
}
/**

6
spring-messaging/src/main/java/org/springframework/messaging/converter/MarshallingMessageConverter.java
View File

@@ -49,8 +49,10 @@ import org.springframework.util.MimeType;
*/
public class MarshallingMessageConverter extends AbstractMessageConverter {
@Nullable
private Marshaller marshaller;
@Nullable
private Unmarshaller unmarshaller;
@@ -97,6 +99,7 @@ public class MarshallingMessageConverter extends AbstractMessageConverter {
/**
* Return the configured Marshaller.
*/
@Nullable
public Marshaller getMarshaller() {
return this.marshaller;
}
@@ -111,6 +114,7 @@ public class MarshallingMessageConverter extends AbstractMessageConverter {
/**
* Return the configured unmarshaller.
*/
@Nullable
public Unmarshaller getUnmarshaller() {
return this.unmarshaller;
}
@@ -123,7 +127,7 @@ public class MarshallingMessageConverter extends AbstractMessageConverter {
}
@Override
protected boolean canConvertTo(Object payload, MessageHeaders headers) {
protected boolean canConvertTo(Object payload, @Nullable MessageHeaders headers) {
return (supportsMimeType(headers) && this.marshaller != null &&
this.marshaller.supports(payload.getClass()));
}

5
spring-messaging/src/main/java/org/springframework/messaging/core/AbstractDestinationResolvingMessagingTemplate.java
View File

@@ -40,7 +40,8 @@ public abstract class AbstractDestinationResolvingMessagingTemplate<D> extends A
DestinationResolvingMessageReceivingOperations<D>,
DestinationResolvingMessageRequestReplyOperations<D> {
private volatile DestinationResolver<D> destinationResolver;
@Nullable
private DestinationResolver<D> destinationResolver;
/**
@@ -58,6 +59,7 @@ public abstract class AbstractDestinationResolvingMessagingTemplate<D> extends A
/**
* Return the configured destination resolver.
*/
@Nullable
public DestinationResolver<D> getDestinationResolver() {
return this.destinationResolver;
}
@@ -70,6 +72,7 @@ public abstract class AbstractDestinationResolvingMessagingTemplate<D> extends A
}
protected final D resolveDestination(String destinationName) {
Assert.state(this.destinationResolver != null, "DestinationResolver is required to resolve destination names");
return this.destinationResolver.resolveDestination(destinationName);
}

5
spring-messaging/src/main/java/org/springframework/messaging/core/AbstractMessageSendingTemplate.java
View File

@@ -52,9 +52,10 @@ public abstract class AbstractMessageSendingTemplate<D> implements MessageSendin
protected final Log logger = LogFactory.getLog(getClass());
private volatile D defaultDestination;
@Nullable
private D defaultDestination;
private volatile MessageConverter converter = new SimpleMessageConverter();
private MessageConverter converter = new SimpleMessageConverter();
/**

4
spring-messaging/src/main/java/org/springframework/messaging/core/BeanFactoryMessageChannelDestinationResolver.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2017 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.
@@ -19,6 +19,7 @@ package org.springframework.messaging.core;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
@@ -33,6 +34,7 @@ import org.springframework.util.Assert;
public class BeanFactoryMessageChannelDestinationResolver
implements DestinationResolver<MessageChannel>, BeanFactoryAware {
@Nullable
private BeanFactory beanFactory;

5
spring-messaging/src/main/java/org/springframework/messaging/core/CachingDestinationResolverProxy.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -20,6 +20,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
@@ -37,6 +38,7 @@ public class CachingDestinationResolverProxy<D> implements DestinationResolver<D
private final Map<String, D> resolvedDestinationCache = new ConcurrentHashMap<>();
@Nullable
private DestinationResolver<D> targetDestinationResolver;
@@ -85,6 +87,7 @@ public class CachingDestinationResolverProxy<D> implements DestinationResolver<D
public D resolveDestination(String name) throws DestinationResolutionException {
D destination = this.resolvedDestinationCache.get(name);
if (destination == null) {
Assert.state(this.targetDestinationResolver != null, "No target DestinationResolver set");
destination = this.targetDestinationResolver.resolveDestination(name);
this.resolvedDestinationCache.put(name, destination);
}

14
spring-messaging/src/main/java/org/springframework/messaging/handler/HandlerMethod.java
View File

@@ -55,6 +55,7 @@ public class HandlerMethod {
private final Object bean;
@Nullable
private final BeanFactory beanFactory;
private final Class<?> beanType;
@@ -65,6 +66,7 @@ public class HandlerMethod {
private final MethodParameter[] parameters;
@Nullable
private HandlerMethod resolvedFromHandlerMethod;
@@ -109,7 +111,10 @@ public class HandlerMethod {
this.bean = beanName;
this.beanFactory = beanFactory;
Class<?> beanType = beanFactory.getType(beanName);
this.beanType = (beanType != null ? ClassUtils.getUserClass(beanType) : null);
if (beanType == null) {
throw new IllegalStateException("Cannot resolve bean type for bean with name '" + beanName + "'");
}
this.beanType = ClassUtils.getUserClass(beanType);
this.method = method;
this.bridgedMethod = BridgeMethodResolver.findBridgedMethod(method);
this.parameters = initMethodParameters();
@@ -204,7 +209,7 @@ public class HandlerMethod {
/**
* Return the actual return value type.
*/
public MethodParameter getReturnValueType(Object returnValue) {
public MethodParameter getReturnValueType(@Nullable Object returnValue) {
return new ReturnValueMethodParameter(returnValue);
}
@@ -244,6 +249,7 @@ public class HandlerMethod {
* resolved via {@link #createWithResolvedBean()}.
* @since 4.3
*/
@Nullable
public HandlerMethod getResolvedFromHandlerMethod() {
return this.resolvedFromHandlerMethod;
}
@@ -255,6 +261,7 @@ public class HandlerMethod {
public HandlerMethod createWithResolvedBean() {
Object handler = this.bean;
if (this.bean instanceof String) {
Assert.state(this.beanFactory != null, "Cannot resolve bean name without BeanFactory");
String beanName = (String) this.bean;
handler = this.beanFactory.getBean(beanName);
}
@@ -333,9 +340,10 @@ public class HandlerMethod {
*/
private class ReturnValueMethodParameter extends HandlerMethodParameter {
@Nullable
private final Object returnValue;
public ReturnValueMethodParameter(Object returnValue) {
public ReturnValueMethodParameter(@Nullable Object returnValue) {
super(-1);
this.returnValue = returnValue;
}

15
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/AbstractExceptionHandlerMethodResolver.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -21,12 +21,11 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.core.ExceptionDepthComparator;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ConcurrentReferenceHashMap;
/**
* Cache exception handling method mappings and provide options to look up a method
@@ -39,11 +38,9 @@ import org.springframework.util.ClassUtils;
*/
public abstract class AbstractExceptionHandlerMethodResolver {
private static final Method NO_METHOD_FOUND = ClassUtils.getMethodIfAvailable(System.class, "currentTimeMillis");
private final Map<Class<? extends Throwable>, Method> mappedMethods = new ConcurrentReferenceHashMap<>(16);
private final Map<Class<? extends Throwable>, Method> mappedMethods = new ConcurrentHashMap<>(16);
private final Map<Class<? extends Throwable>, Method> exceptionLookupCache = new ConcurrentHashMap<>(16);
private final Map<Class<? extends Throwable>, Method> exceptionLookupCache = new ConcurrentReferenceHashMap<>(16);
/**
@@ -109,9 +106,9 @@ public abstract class AbstractExceptionHandlerMethodResolver {
Method method = this.exceptionLookupCache.get(exceptionType);
if (method == null) {
method = getMappedMethod(exceptionType);
this.exceptionLookupCache.put(exceptionType, method != null ? method : NO_METHOD_FOUND);
this.exceptionLookupCache.put(exceptionType, method);
}
return method != NO_METHOD_FOUND ? method : null;
return method;
}
/**

1
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/InvocableHandlerMethod.java
View File

@@ -259,6 +259,7 @@ public class InvocableHandlerMethod extends HandlerMethod {
private class AsyncResultMethodParameter extends HandlerMethodParameter {
@Nullable
private final Object returnValue;
private final ResolvableType returnType;

4
spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/support/SendToMethodReturnValueHandler.java
View File

@@ -69,11 +69,12 @@ public class SendToMethodReturnValueHandler implements HandlerMethodReturnValueH
private PropertyPlaceholderHelper placeholderHelper = new PropertyPlaceholderHelper("{", "}", null, false);
@Nullable
private MessageHeaderInitializer headerInitializer;
public SendToMethodReturnValueHandler(SimpMessageSendingOperations messagingTemplate, boolean annotationRequired) {
Assert.notNull(messagingTemplate, "messagingTemplate must not be null");
Assert.notNull(messagingTemplate, "'messagingTemplate' must not be null");
this.messagingTemplate = messagingTemplate;
this.annotationRequired = annotationRequired;
}
@@ -276,6 +277,7 @@ public class SendToMethodReturnValueHandler implements HandlerMethodReturnValueH
private static class DestinationVariablePlaceholderResolver implements PlaceholderResolver {
@Nullable
private final Map<String, String> vars;
public DestinationVariablePlaceholderResolver(@Nullable Map<String, String> vars) {

9
spring-messaging/src/main/java/org/springframework/messaging/simp/annotation/support/SimpAnnotationMethodMessageHandler.java
View File

@@ -101,10 +101,13 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
private boolean slashPathSeparator = true;
@Nullable
private Validator validator;
@Nullable
private StringValueResolver valueResolver;
@Nullable
private MessageHeaderInitializer headerInitializer;
private final Object lifecycleMonitor = new Object();
@@ -221,6 +224,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
/**
* Return the configured Validator instance.
*/
@Nullable
public Validator getValidator() {
return this.validator;
}
@@ -252,6 +256,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
/**
* Return the configured header initializer.
*/
@Nullable
public MessageHeaderInitializer getHeaderInitializer() {
return this.headerInitializer;
}
@@ -331,7 +336,9 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
// Annotation-based return value types
SendToMethodReturnValueHandler sendToHandler =
new SendToMethodReturnValueHandler(this.brokerTemplate, true);
sendToHandler.setHeaderInitializer(this.headerInitializer);
if (this.headerInitializer != null) {
sendToHandler.setHeaderInitializer(this.headerInitializer);
}
handlers.add(sendToHandler);
SubscriptionMethodReturnValueHandler subscriptionHandler =

2
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/AbstractBrokerMessageHandler.java
View File

@@ -59,6 +59,7 @@ public abstract class AbstractBrokerMessageHandler
private final Collection<String> destinationPrefixes;
@Nullable
private ApplicationEventPublisher eventPublisher;
private AtomicBoolean brokerAvailable = new AtomicBoolean(false);
@@ -132,6 +133,7 @@ public abstract class AbstractBrokerMessageHandler
this.eventPublisher = publisher;
}
@Nullable
public ApplicationEventPublisher getApplicationEventPublisher() {
return this.eventPublisher;
}

1
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/DefaultSubscriptionRegistry.java
View File

@@ -492,6 +492,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
private final String id;
@Nullable
private final Expression selectorExpression;
public Subscription(String id, @Nullable Expression selector) {

19
spring-messaging/src/main/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandler.java
View File

@@ -56,16 +56,22 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
private SubscriptionRegistry subscriptionRegistry;
@Nullable
private PathMatcher pathMatcher;
@Nullable
private Integer cacheLimit;
@Nullable
private TaskScheduler taskScheduler;
@Nullable
private long[] heartbeatValue;
@Nullable
private ScheduledFuture<?> heartbeatFuture;
@Nullable
private MessageHeaderInitializer headerInitializer;
@@ -112,7 +118,7 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
* @see DefaultSubscriptionRegistry#setPathMatcher
* @see org.springframework.util.AntPathMatcher
*/
public void setPathMatcher(PathMatcher pathMatcher) {
public void setPathMatcher(@Nullable PathMatcher pathMatcher) {
this.pathMatcher = pathMatcher;
initPathMatcherToUse();
}
@@ -133,7 +139,7 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
* @see DefaultSubscriptionRegistry#setCacheLimit
* @see DefaultSubscriptionRegistry#DEFAULT_CACHE_LIMIT
*/
public void setCacheLimit(Integer cacheLimit) {
public void setCacheLimit(@Nullable Integer cacheLimit) {
this.cacheLimit = cacheLimit;
initCacheLimitToUse();
}
@@ -216,7 +222,7 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
@Override
public void startInternal() {
publishBrokerAvailableEvent();
if (getTaskScheduler() != null) {
if (this.taskScheduler != null) {
long interval = initHeartbeatTaskDelay();
if (interval > 0) {
this.heartbeatFuture = this.taskScheduler.scheduleWithFixedDelay(new HeartbeatTask(), interval);
@@ -384,6 +390,7 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
private final String sessiondId;
@Nullable
private final Principal user;
private final long readInterval;
@@ -416,6 +423,7 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
return this.sessiondId;
}
@Nullable
public Principal getUser() {
return this.user;
}
@@ -458,7 +466,10 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
if (info.getWriteInterval() > 0 && (now - info.getLastWriteTime()) > info.getWriteInterval()) {
SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.create(SimpMessageType.HEARTBEAT);
accessor.setSessionId(info.getSessiondId());
accessor.setUser(info.getUser());
Principal user = info.getUser();
if (user != null) {
accessor.setUser(user);
}
initHeaders(accessor);
MessageHeaders headers = accessor.getMessageHeaders();
getClientOutboundChannel().send(MessageBuilder.createMessage(EMPTY_PAYLOAD, headers));

7
spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java
View File

@@ -91,12 +91,16 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
"com.fasterxml.jackson.databind.ObjectMapper", AbstractMessageBrokerConfiguration.class.getClassLoader());
@Nullable
private ApplicationContext applicationContext;
@Nullable
private ChannelRegistration clientInboundChannelRegistration;
@Nullable
private ChannelRegistration clientOutboundChannelRegistration;
@Nullable
private MessageBrokerRegistry brokerRegistry;
@@ -112,6 +116,7 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
this.applicationContext = applicationContext;
}
@Nullable
public ApplicationContext getApplicationContext() {
return this.applicationContext;
}
@@ -418,7 +423,7 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
protected Validator simpValidator() {
Validator validator = getValidator();
if (validator == null) {
if (this.applicationContext.containsBean(MVC_VALIDATOR_NAME)) {
if (this.applicationContext != null && this.applicationContext.containsBean(MVC_VALIDATOR_NAME)) {
validator = this.applicationContext.getBean(MVC_VALIDATOR_NAME, Validator.class);
}
else if (ClassUtils.isPresent("javax.validation.Validator", getClass().getClassLoader())) {

5
spring-messaging/src/main/java/org/springframework/messaging/simp/config/ChannelRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.7
* Copyright 2002-2017 the original author or authors.7
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -20,6 +20,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.springframework.lang.Nullable;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@@ -32,6 +33,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
*/
public class ChannelRegistration {
@Nullable
private TaskExecutorRegistration registration;
private final List<ChannelInterceptor> interceptors = new ArrayList<>();
@@ -71,6 +73,7 @@ public class ChannelRegistration {
return (this.registration != null);
}
@Nullable
protected TaskExecutorRegistration getTaskExecRegistration() {
return this.registration;
}

6
spring-messaging/src/main/java/org/springframework/messaging/simp/config/MessageBrokerRegistry.java
View File

@@ -40,18 +40,24 @@ public class MessageBrokerRegistry {
private final MessageChannel clientOutboundChannel;
@Nullable
private SimpleBrokerRegistration simpleBrokerRegistration;
@Nullable
private StompBrokerRelayRegistration brokerRelayRegistration;
private final ChannelRegistration brokerChannelRegistration = new ChannelRegistration();
@Nullable
private String[] applicationDestinationPrefixes;
@Nullable
private String userDestinationPrefix;
@Nullable
private PathMatcher pathMatcher;
@Nullable
private Integer cacheLimit;

5
spring-messaging/src/main/java/org/springframework/messaging/simp/config/SimpleBrokerRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2017 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.
@@ -16,6 +16,7 @@
package org.springframework.messaging.simp.config;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler;
@@ -29,8 +30,10 @@ import org.springframework.scheduling.TaskScheduler;
*/
public class SimpleBrokerRegistration extends AbstractBrokerRegistration {
@Nullable
private TaskScheduler taskScheduler;
@Nullable
private long[] heartbeat;

10
spring-messaging/src/main/java/org/springframework/messaging/simp/config/StompBrokerRelayRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2017 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.
@@ -16,6 +16,7 @@
package org.springframework.messaging.simp.config;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler;
@@ -41,16 +42,21 @@ public class StompBrokerRelayRegistration extends AbstractBrokerRegistration {
private String systemPasscode = "guest";
@Nullable
private Long systemHeartbeatSendInterval;
@Nullable
private Long systemHeartbeatReceiveInterval;
@Nullable
private String virtualHost;
private boolean autoStartup = true;
@Nullable
private String userDestinationBroadcast;
@Nullable
private String userRegistryBroadcast;
@@ -184,6 +190,7 @@ public class StompBrokerRelayRegistration extends AbstractBrokerRegistration {
return this;
}
@Nullable
protected String getUserDestinationBroadcast() {
return this.userDestinationBroadcast;
}
@@ -202,6 +209,7 @@ public class StompBrokerRelayRegistration extends AbstractBrokerRegistration {
return this;
}
@Nullable
protected String getUserRegistryBroadcast() {
return this.userRegistryBroadcast;
}

4
spring-messaging/src/main/java/org/springframework/messaging/simp/config/TaskExecutorRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2017 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.
@@ -16,6 +16,7 @@
package org.springframework.messaging.simp.config;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
/**
@@ -26,6 +27,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
*/
public class TaskExecutorRegistration {
@Nullable
private ThreadPoolTaskExecutor taskExecutor;
private int corePoolSize = Runtime.getRuntime().availableProcessors() * 2;

13
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/BufferingStompDecoder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.LinkedMultiValueMap;
@@ -53,6 +54,7 @@ public class BufferingStompDecoder {
private final Queue<ByteBuffer> chunks = new LinkedBlockingQueue<>();
@Nullable
private volatile Integer expectedContentLength;
@@ -102,7 +104,8 @@ public class BufferingStompDecoder {
this.chunks.add(newBuffer);
checkBufferLimits();
if (this.expectedContentLength != null && getBufferSize() < this.expectedContentLength) {
Integer contentLength = this.expectedContentLength;
if (contentLength != null && getBufferSize() < contentLength) {
return Collections.emptyList();
}
@@ -136,8 +139,9 @@ public class BufferingStompDecoder {
}
private void checkBufferLimits() {
if (this.expectedContentLength != null) {
if (this.expectedContentLength > this.bufferSizeLimit) {
Integer contentLength = this.expectedContentLength;
if (contentLength != null) {
if (contentLength > this.bufferSizeLimit) {
throw new StompConversionException(
"STOMP 'content-length' header value " + this.expectedContentLength +
" exceeds configured buffer size limit " + this.bufferSizeLimit);
@@ -163,6 +167,7 @@ public class BufferingStompDecoder {
/**
* Get the expected content length of the currently buffered, incomplete STOMP frame.
*/
@Nullable
public Integer getExpectedContentLength() {
return this.expectedContentLength;
}

17
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/DefaultStompSession.java
View File

@@ -84,6 +84,7 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
private MessageConverter converter = new SimpleMessageConverter();
@Nullable
private TaskScheduler taskScheduler;
private long receiptTimeLimit = 15 * 1000;
@@ -91,8 +92,10 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
private volatile boolean autoReceiptEnabled;
@Nullable
private volatile TcpConnection<byte[]> connection;
@Nullable
private volatile String version;
private final AtomicInteger subscriptionIndex = new AtomicInteger();
@@ -167,6 +170,7 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
/**
* Return the configured TaskScheduler to use for receipt tracking.
*/
@Nullable
public TaskScheduler getTaskScheduler() {
return this.taskScheduler;
}
@@ -464,13 +468,15 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
if (connect == null || connected == null) {
return;
}
TcpConnection<byte[]> con = this.connection;
Assert.state(con != null, "No TcpConnection available");
if (connect[0] > 0 && connected[1] > 0) {
long interval = Math.max(connect[0], connected[1]);
this.connection.onWriteInactivity(new WriteInactivityTask(), interval);
con.onWriteInactivity(new WriteInactivityTask(), interval);
}
if (connect[1] > 0 && connected[0] > 0) {
final long interval = Math.max(connect[1], connected[0]) * HEARTBEAT_MULTIPLIER;
this.connection.onReadInactivity(new ReadInactivityTask(), interval);
long interval = Math.max(connect[1], connected[0]) * HEARTBEAT_MULTIPLIER;
con.onReadInactivity(new ReadInactivityTask(), interval);
}
}
@@ -514,14 +520,17 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
private class ReceiptHandler implements Receiptable {
@Nullable
private final String receiptId;
private final List<Runnable> receiptCallbacks = new ArrayList<>(2);
private final List<Runnable> receiptLostCallbacks = new ArrayList<>(2);
@Nullable
private ScheduledFuture<?> future;
@Nullable
private Boolean result;
public ReceiptHandler(@Nullable String receiptId) {
@@ -666,7 +675,7 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
if (conn != null) {
conn.send(HEARTBEAT).addCallback(
new ListenableFutureCallback<Void>() {
public void onSuccess(Void result) {
public void onSuccess(@Nullable Void result) {
}
public void onFailure(Throwable ex) {
handleFailure(ex);

62
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java
View File

@@ -123,10 +123,13 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
private final Map<String, MessageHandler> systemSubscriptions = new HashMap<>(4);
@Nullable
private String virtualHost;
@Nullable
private TcpOperations<byte[]> tcpClient;
@Nullable
private MessageHeaderInitializer headerInitializer;
private final Stats stats = new Stats();
@@ -389,7 +392,9 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
protected void startInternal() {
if (this.tcpClient == null) {
StompDecoder decoder = new StompDecoder();
decoder.setHeaderInitializer(getHeaderInitializer());
if (this.headerInitializer != null) {
decoder.setHeaderInitializer(this.headerInitializer);
}
ReactorNettyCodec<byte[]> codec = new StompReactorNettyCodec(decoder);
this.tcpClient = new ReactorNettyTcpClient<>(this.relayHost, this.relayPort, codec);
}
@@ -422,11 +427,13 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
@Override
protected void stopInternal() {
publishBrokerUnavailableEvent();
try {
this.tcpClient.shutdown().get(5000, TimeUnit.MILLISECONDS);
}
catch (Throwable ex) {
logger.error("Error in shutdown of TCP client", ex);
if (this.tcpClient != null) {
try {
this.tcpClient.shutdown().get(5000, TimeUnit.MILLISECONDS);
}
catch (Throwable ex) {
logger.error("Error in shutdown of TCP client", ex);
}
}
}
@@ -514,6 +521,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
StompConnectionHandler handler = new StompConnectionHandler(sessionId, stompAccessor);
this.connectionHandlers.put(sessionId, handler);
this.stats.incrementConnectCount();
Assert.state(this.tcpClient != null, "No TCP client available");
this.tcpClient.connect(handler);
}
else if (StompCommand.DISCONNECT.equals(command)) {
@@ -553,6 +561,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
private final StompHeaderAccessor connectHeaders;
@Nullable
private volatile TcpConnection<byte[]> tcpConnection;
private volatile boolean isStompConnected;
@@ -585,7 +594,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
logger.debug("TCP connection opened in session=" + getSessionId());
}
this.tcpConnection = connection;
this.tcpConnection.onReadInactivity(() -> {
connection.onReadInactivity(() -> {
if (this.tcpConnection != null && !this.isStompConnected) {
handleTcpConnectionFailure("No CONNECTED frame received in " +
MAX_TIME_TO_CONNECTED_FRAME + " ms.", null);
@@ -691,6 +700,9 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
return;
}
TcpConnection<byte[]> con = this.tcpConnection;
Assert.state(con != null, "No TcpConnection available");
long clientSendInterval = this.connectHeaders.getHeartbeat()[0];
long clientReceiveInterval = this.connectHeaders.getHeartbeat()[1];
long serverSendInterval = connectedHeaders.getHeartbeat()[0];
@@ -698,25 +710,16 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
if (clientSendInterval > 0 && serverReceiveInterval > 0) {
long interval = Math.max(clientSendInterval, serverReceiveInterval);
this.tcpConnection.onWriteInactivity(() -> {
TcpConnection<byte[]> conn = this.tcpConnection;
if (conn != null) {
conn.send(HEARTBEAT_MESSAGE).addCallback(
new ListenableFutureCallback<Void>() {
public void onSuccess(Void result) {
}
public void onFailure(Throwable ex) {
handleTcpConnectionFailure("Failed to forward heartbeat: " + ex.getMessage(), ex);
}
});
}
}, interval);
con.onWriteInactivity(() ->
con.send(HEARTBEAT_MESSAGE).addCallback(
result -> {},
ex -> handleTcpConnectionFailure(
"Failed to forward heartbeat: " + ex.getMessage(), ex)), interval);
}
if (clientReceiveInterval > 0 && serverSendInterval > 0) {
final long interval = Math.max(clientReceiveInterval, serverSendInterval) * HEARTBEAT_MULTIPLIER;
this.tcpConnection.onReadInactivity(()
-> handleTcpConnectionFailure("No messages received in " + interval + " ms.", null), interval);
con.onReadInactivity(
() -> handleTcpConnectionFailure("No messages received in " + interval + " ms.", null), interval);
}
}
@@ -809,7 +812,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
ListenableFuture<Void> future = conn.send((Message<byte[]>) messageToSend);
future.addCallback(new ListenableFutureCallback<Void>() {
@Override
public void onSuccess(Void result) {
public void onSuccess(@Nullable Void result) {
if (accessor.getCommand() == StompCommand.DISCONNECT) {
afterDisconnectSent(accessor);
}
@@ -910,13 +913,10 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
if (conn != null) {
MessageHeaders headers = accessor.getMessageHeaders();
conn.send(MessageBuilder.createMessage(EMPTY_PAYLOAD, headers)).addCallback(
new ListenableFutureCallback<Void>() {
public void onSuccess(Void result) {
}
public void onFailure(Throwable ex) {
String error = "Failed to subscribe in \"system\" session.";
handleTcpConnectionFailure(error, ex);
}
result -> {},
ex -> {
String error = "Failed to subscribe in \"system\" session.";
handleTcpConnectionFailure(error, ex);
});
}
}

2
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompClientSupport.java
View File

@@ -71,7 +71,7 @@ public abstract class StompClientSupport {
/**
* Configure a scheduler to use for heartbeats and for receipt tracking.
* <p><strong>Note:</strong> some transports have built-in support to work
* <p><strong>Note:</strong> Some transports have built-in support to work
* with heartbeats and therefore do not require a TaskScheduler.
* Receipts however, if needed, do require a TaskScheduler to be configured.
* <p>By default, this is not set.

3
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompDecoder.java
View File

@@ -53,6 +53,7 @@ public class StompDecoder {
private static final Log logger = LogFactory.getLog(StompDecoder.class);
@Nullable
private MessageHeaderInitializer headerInitializer;
@@ -60,7 +61,7 @@ public class StompDecoder {
* Configure a {@link MessageHeaderInitializer} to apply to the headers of
* {@link Message}s from decoded STOMP frames.
*/
public void setHeaderInitializer(@Nullable MessageHeaderInitializer headerInitializer) {
public void setHeaderInitializer(MessageHeaderInitializer headerInitializer) {
this.headerInitializer = headerInitializer;
}

3
spring-messaging/src/main/java/org/springframework/messaging/simp/user/DefaultUserDestinationResolver.java
View File

@@ -207,7 +207,7 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver {
Set<String> sessionIds;
SimpUser user = this.userRegistry.getUser(userName);
if (user != null) {
if (user.getSession(sessionId) != null) {
if (sessionId != null && user.getSession(sessionId) != null) {
sessionIds = Collections.singleton(sessionId);
}
else {
@@ -264,6 +264,7 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver {
private final Set<String> sessionIds;
@Nullable
private final String user;
public ParseResult(String sourceDest, String actualDest, String subscribeDest,

4
spring-messaging/src/main/java/org/springframework/messaging/simp/user/MultiServerUserRegistry.java
View File

@@ -29,6 +29,7 @@ import java.util.concurrent.ConcurrentHashMap;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.event.SmartApplicationListener;
import org.springframework.core.Ordered;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
@@ -97,7 +98,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
}
@Override
public boolean supportsSourceType(Class<?> sourceType) {
public boolean supportsSourceType(@Nullable Class<?> sourceType) {
return (this.delegateApplicationEvents &&
((SmartApplicationListener) this.localRegistry).supportsSourceType(sourceType));
}
@@ -278,6 +279,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
private Set<TransferSimpSession> sessions;
/* Cross-server session lookup (e.g. user connected to multiple servers) */
@Nullable
private SessionLookup sessionLookup;
/**

8
spring-messaging/src/main/java/org/springframework/messaging/simp/user/SimpSubscription.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2017 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.
@@ -25,17 +25,17 @@ package org.springframework.messaging.simp.user;
public interface SimpSubscription {
/**
* Return the id associated of the subscription, never {@code null}.
* Return the id associated of the subscription (never {@code null}).
*/
String getId();
/**
* Return the session of the subscription, never {@code null}.
* Return the session of the subscription (never {@code null}).
*/
SimpSession getSession();
/**
* Return the subscription's destination, never {@code null}.
* Return the subscription's destination (never {@code null}).
*/
String getDestination();

2
spring-messaging/src/main/java/org/springframework/messaging/simp/user/SimpUser.java
View File

@@ -44,7 +44,7 @@ public interface SimpUser {
* @return the matching session, or {@code null} if none found
*/
@Nullable
SimpSession getSession(@Nullable String sessionId);
SimpSession getSession(String sessionId);
/**
* Return the sessions for the user.

2
spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationMessageHandler.java
View File

@@ -61,8 +61,10 @@ public class UserDestinationMessageHandler implements MessageHandler, SmartLifec
private final MessageSendingOperations<String> messagingTemplate;
@Nullable
private BroadcastHandler broadcastHandler;
@Nullable
private MessageHeaderInitializer headerInitializer;
private final Object lifecycleMonitor = new Object();

1
spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserDestinationResult.java
View File

@@ -37,6 +37,7 @@ public class UserDestinationResult {
private final String subscribeDestination;
@Nullable
private final String user;

20
spring-messaging/src/main/java/org/springframework/messaging/simp/user/UserRegistryMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -19,6 +19,7 @@ package org.springframework.messaging.simp.user;
import java.util.concurrent.ScheduledFuture;
import org.springframework.context.ApplicationListener;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
@@ -35,8 +36,7 @@ import org.springframework.util.Assert;
* application servers and periodically broadcasts the content of the local
* user registry.
*
* The aggregated information
* is maintained in a {@link MultiServerUserRegistry}.
* <p>The aggregated information is maintained in a {@link MultiServerUserRegistry}.
*
* @author Rossen Stoyanchev
* @since 4.2
@@ -53,6 +53,7 @@ public class UserRegistryMessageHandler implements MessageHandler, ApplicationLi
private final UserRegistryTask schedulerTask = new UserRegistryTask();
@Nullable
private volatile ScheduledFuture<?> scheduledFuture;
private long registryExpirationPeriod = 20 * 1000;
@@ -62,8 +63,8 @@ public class UserRegistryMessageHandler implements MessageHandler, ApplicationLi
* Constructor.
* @param userRegistry the registry with local and remote user registry information
* @param brokerTemplate template for broadcasting local registry information
* @param broadcastDestination the destination to broadcast to
* @param scheduler
* @param broadcastDestination the destination to broadcast to
* @param scheduler the task scheduler to use
*/
public UserRegistryMessageHandler(MultiServerUserRegistry userRegistry,
SimpMessagingTemplate brokerTemplate, String broadcastDestination, TaskScheduler scheduler) {
@@ -112,9 +113,12 @@ public class UserRegistryMessageHandler implements MessageHandler, ApplicationLi
long delay = getRegistryExpirationPeriod() / 2;
this.scheduledFuture = this.scheduler.scheduleWithFixedDelay(this.schedulerTask, delay);
}
else if (this.scheduledFuture != null ){
this.scheduledFuture.cancel(true);
this.scheduledFuture = null;
else {
ScheduledFuture<?> future = this.scheduledFuture;
if (future != null ){
future.cancel(true);
this.scheduledFuture = null;
}
}
}

8
spring-messaging/src/main/java/org/springframework/messaging/support/ErrorMessage.java
View File

@@ -18,6 +18,7 @@ package org.springframework.messaging.support;
import java.util.Map;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
@@ -44,6 +45,7 @@ public class ErrorMessage extends GenericMessage<Throwable> {
private static final long serialVersionUID = -5470210965279837728L;
@Nullable
private final Message<?> originalMessage;
@@ -126,6 +128,7 @@ public class ErrorMessage extends GenericMessage<Throwable> {
* where the ErrorMessage was created.
* @since 5.0
*/
@Nullable
public Message<?> getOriginalMessage() {
return this.originalMessage;
}
@@ -135,10 +138,7 @@ public class ErrorMessage extends GenericMessage<Throwable> {
if (this.originalMessage == null) {
return super.toString();
}
StringBuilder sb = new StringBuilder(super.toString());
sb.append(" for original ").append(this.originalMessage);
return sb.toString();
return super.toString() + " for original " + this.originalMessage;
}
}

2
spring-messaging/src/main/java/org/springframework/messaging/support/ExecutorSubscribableChannel.java
View File

@@ -36,6 +36,7 @@ import org.springframework.messaging.SubscribableChannel;
*/
public class ExecutorSubscribableChannel extends AbstractSubscribableChannel {
@Nullable
private final Executor executor;
private final List<ExecutorChannelInterceptor> executorInterceptors = new ArrayList<>(4);
@@ -60,6 +61,7 @@ public class ExecutorSubscribableChannel extends AbstractSubscribableChannel {
}
@Nullable
public Executor getExecutor() {
return this.executor;
}

1
spring-messaging/src/main/java/org/springframework/messaging/support/MessageBuilder.java
View File

@@ -39,6 +39,7 @@ public final class MessageBuilder<T> {
private final T payload;
@Nullable
private final Message<T> originalMessage;
private MessageHeaderAccessor headerAccessor;

1
spring-messaging/src/main/java/org/springframework/messaging/support/MessageHeaderAccessor.java
View File

@@ -130,6 +130,7 @@ public class MessageHeaderAccessor {
private boolean enableTimestamp = false;
@Nullable
private IdGenerator idGenerator;