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

Explicit type can be replaced by <>

Browse Source 
Issue: SPR-13188
This commit is contained in:
Stephane Nicoll
2016-07-05 17:00:26 +02:00
parent 3096888c7d
commit 00d2606b00
1044 changed files with 3972 additions and 3893 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -117,7 +117,7 @@ public class MessageHeaders implements Map<String, Object>, Serializable {
* @param timestamp the {@link #TIMESTAMP} header value
*/
protected MessageHeaders(Map<String, Object> headers, UUID id, Long timestamp) {
this.headers = (headers != null ? new HashMap<String, Object>(headers) : new HashMap<String, Object>());
this.headers = (headers != null ? new HashMap<>(headers) : new HashMap<String, Object>());
if (id == null) {
this.headers.put(ID, getIdGenerator().generateId());
@@ -147,7 +147,7 @@ public class MessageHeaders implements Map<String, Object>, Serializable {
* @param keysToIgnore the keys of the entries to ignore
*/
private MessageHeaders(MessageHeaders original, Set<String> keysToIgnore) {
this.headers = new HashMap<String, Object>(original.headers.size() - keysToIgnore.size());
this.headers = new HashMap<>(original.headers.size() - keysToIgnore.size());
for (Map.Entry<String, Object> entry : original.headers.entrySet()) {
if (!keysToIgnore.contains(entry.getKey())) {
this.headers.put(entry.getKey(), entry.getValue());
@@ -268,7 +268,7 @@ public class MessageHeaders implements Map<String, Object>, Serializable {
// Serialization methods
private void writeObject(ObjectOutputStream out) throws IOException {
Set<String> keysToIgnore = new HashSet<String>();
Set<String> keysToIgnore = new HashSet<>();
for (Map.Entry<String, Object> entry : this.headers.entrySet()) {
if (!(entry.getValue() instanceof Serializable)) {
keysToIgnore.add(entry.getKey());

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

@@ -70,7 +70,7 @@ public abstract class AbstractMessageConverter implements SmartMessageConverter
*/
protected AbstractMessageConverter(Collection<MimeType> supportedMimeTypes) {
Assert.notNull(supportedMimeTypes, "supportedMimeTypes must not be null");
this.supportedMimeTypes = new ArrayList<MimeType>(supportedMimeTypes);
this.supportedMimeTypes = new ArrayList<>(supportedMimeTypes);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -45,7 +45,7 @@ public class CompositeMessageConverter implements SmartMessageConverter {
*/
public CompositeMessageConverter(Collection<MessageConverter> converters) {
Assert.notEmpty(converters, "Converters must not be empty");
this.converters = new ArrayList<MessageConverter>(converters);
this.converters = new ArrayList<>(converters);
}

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

@@ -145,7 +145,7 @@ public class MappingJackson2MessageConverter extends AbstractMessageConverter {
if (!logger.isWarnEnabled()) {
return this.objectMapper.canDeserialize(javaType);
}
AtomicReference<Throwable> causeRef = new AtomicReference<Throwable>();
AtomicReference<Throwable> causeRef = new AtomicReference<>();
if (this.objectMapper.canDeserialize(javaType, causeRef)) {
return true;
}
@@ -161,7 +161,7 @@ public class MappingJackson2MessageConverter extends AbstractMessageConverter {
if (!logger.isWarnEnabled()) {
return this.objectMapper.canSerialize(payload.getClass());
}
AtomicReference<Throwable> causeRef = new AtomicReference<Throwable>();
AtomicReference<Throwable> causeRef = new AtomicReference<>();
if (this.objectMapper.canSerialize(payload.getClass(), causeRef)) {
return true;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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,7 +35,7 @@ import org.springframework.util.Assert;
*/
public class CachingDestinationResolverProxy<D> implements DestinationResolver<D>, InitializingBean {
private final Map<String, D> resolvedDestinationCache = new ConcurrentHashMap<String, D>();
private final Map<String, D> resolvedDestinationCache = new ConcurrentHashMap<>();
private DestinationResolver<D> targetDestinationResolver;

8
spring-messaging/src/main/java/org/springframework/messaging/handler/DestinationPatternsMessageCondition.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -81,7 +81,7 @@ public class DestinationPatternsMessageCondition extends AbstractMessageConditio
return Collections.emptySet();
}
boolean slashSeparator = pathMatcher.combine("a", "a").equals("a/a");
Set<String> result = new LinkedHashSet<String>(patterns.size());
Set<String> result = new LinkedHashSet<>(patterns.size());
for (String pattern : patterns) {
if (slashSeparator) {
if (StringUtils.hasLength(pattern) && !pattern.startsWith("/")) {
@@ -121,7 +121,7 @@ public class DestinationPatternsMessageCondition extends AbstractMessageConditio
*/
@Override
public DestinationPatternsMessageCondition combine(DestinationPatternsMessageCondition other) {
Set<String> result = new LinkedHashSet<String>();
Set<String> result = new LinkedHashSet<>();
if (!this.patterns.isEmpty() && !other.patterns.isEmpty()) {
for (String pattern1 : this.patterns) {
for (String pattern2 : other.patterns) {
@@ -161,7 +161,7 @@ public class DestinationPatternsMessageCondition extends AbstractMessageConditio
return this;
}
List<String> matches = new ArrayList<String>();
List<String> matches = new ArrayList<>();
for (String pattern : this.patterns) {
if (pattern.equals(destination) || this.pathMatcher.match(pattern, destination)) {
matches.add(pattern);

2
spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/AbstractNamedValueMethodArgumentResolver.java
View File

@@ -64,7 +64,7 @@ public abstract class AbstractNamedValueMethodArgumentResolver implements Handle
private final BeanExpressionContext expressionContext;
private final Map<MethodParameter, NamedValueInfo> namedValueInfoCache =
new ConcurrentHashMap<MethodParameter, NamedValueInfo>(256);
new ConcurrentHashMap<>(256);
/**

4
spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/AnnotationExceptionHandlerMethodResolver.java
View File

@@ -57,10 +57,10 @@ public class AnnotationExceptionHandlerMethodResolver extends AbstractExceptionH
}
});
Map<Class<? extends Throwable>, Method> result = new HashMap<Class<? extends Throwable>, Method>();
Map<Class<? extends Throwable>, Method> result = new HashMap<>();
for (Map.Entry<Method, MessageExceptionHandler> entry : methods.entrySet()) {
Method method = entry.getKey();
List<Class<? extends Throwable>> exceptionTypes = new ArrayList<Class<? extends Throwable>>();
List<Class<? extends Throwable>> exceptionTypes = new ArrayList<>();
exceptionTypes.addAll(Arrays.asList(entry.getValue().value()));
if (exceptionTypes.isEmpty()) {
exceptionTypes.addAll(getExceptionsFromMethodSignature(method));

2
spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/DefaultMessageHandlerMethodFactory.java
View File

@@ -149,7 +149,7 @@ public class DefaultMessageHandlerMethodFactory implements MessageHandlerMethodF
}
protected List<HandlerMethodArgumentResolver> initArgumentResolvers() {
List<HandlerMethodArgumentResolver> resolvers = new ArrayList<HandlerMethodArgumentResolver>();
List<HandlerMethodArgumentResolver> resolvers = new ArrayList<>();
ConfigurableBeanFactory cbf = (this.beanFactory instanceof ConfigurableBeanFactory ?
(ConfigurableBeanFactory) this.beanFactory : null);

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

@@ -40,9 +40,9 @@ 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 ConcurrentHashMap<Class<? extends Throwable>, Method>(16);
private final Map<Class<? extends Throwable>, Method> mappedMethods = new ConcurrentHashMap<>(16);
private final Map<Class<? extends Throwable>, Method> exceptionLookupCache = new ConcurrentHashMap<Class<? extends Throwable>, Method>(16);
private final Map<Class<? extends Throwable>, Method> exceptionLookupCache = new ConcurrentHashMap<>(16);
/**
@@ -60,7 +60,7 @@ public abstract class AbstractExceptionHandlerMethodResolver {
*/
@SuppressWarnings("unchecked")
protected static List<Class<? extends Throwable>> getExceptionsFromMethodSignature(Method method) {
List<Class<? extends Throwable>> result = new ArrayList<Class<? extends Throwable>>();
List<Class<? extends Throwable>> result = new ArrayList<>();
for (Class<?> paramType : method.getParameterTypes()) {
if (Throwable.class.isAssignableFrom(paramType)) {
result.add((Class<? extends Throwable>) paramType);
@@ -115,7 +115,7 @@ public abstract class AbstractExceptionHandlerMethodResolver {
* Return the {@link Method} mapped to the given exception type, or {@code null} if none.
*/
private Method getMappedMethod(Class<? extends Throwable> exceptionType) {
List<Class<? extends Throwable>> matches = new ArrayList<Class<? extends Throwable>>();
List<Class<? extends Throwable>> matches = new ArrayList<>();
for (Class<? extends Throwable> mappedException : this.mappedMethods.keySet()) {
if (mappedException.isAssignableFrom(exceptionType)) {
matches.add(mappedException);

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

@@ -83,13 +83,13 @@ public abstract class AbstractMethodMessageHandler<T>
protected final Log logger = LogFactory.getLog(getClass());
private Collection<String> destinationPrefixes = new ArrayList<String>();
private Collection<String> destinationPrefixes = new ArrayList<>();
private final List<HandlerMethodArgumentResolver> customArgumentResolvers =
new ArrayList<HandlerMethodArgumentResolver>(4);
new ArrayList<>(4);
private final List<HandlerMethodReturnValueHandler> customReturnValueHandlers =
new ArrayList<HandlerMethodReturnValueHandler>(4);
new ArrayList<>(4);
private final HandlerMethodArgumentResolverComposite argumentResolvers =
new HandlerMethodArgumentResolverComposite();
@@ -99,15 +99,15 @@ public abstract class AbstractMethodMessageHandler<T>
private ApplicationContext applicationContext;
private final Map<T, HandlerMethod> handlerMethods = new LinkedHashMap<T, HandlerMethod>(64);
private final Map<T, HandlerMethod> handlerMethods = new LinkedHashMap<>(64);
private final MultiValueMap<String, T> destinationLookup = new LinkedMultiValueMap<String, T>(64);
private final MultiValueMap<String, T> destinationLookup = new LinkedMultiValueMap<>(64);
private final Map<Class<?>, AbstractExceptionHandlerMethodResolver> exceptionHandlerCache =
new ConcurrentHashMap<Class<?>, AbstractExceptionHandlerMethodResolver>(64);
new ConcurrentHashMap<>(64);
private final Map<MessagingAdviceBean, AbstractExceptionHandlerMethodResolver> exceptionHandlerAdviceCache =
new LinkedHashMap<MessagingAdviceBean, AbstractExceptionHandlerMethodResolver>(64);
new LinkedHashMap<>(64);
/**

2
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/CompletableFutureReturnValueHandler.java
View File

@@ -38,7 +38,7 @@ public class CompletableFutureReturnValueHandler extends AbstractAsyncReturnValu
@Override
@SuppressWarnings("unchecked")
public ListenableFuture<?> toListenableFuture(Object returnValue, MethodParameter returnType) {
return new CompletableToListenableFutureAdapter<Object>((CompletableFuture<Object>) returnValue);
return new CompletableToListenableFutureAdapter<>((CompletableFuture<Object>) returnValue);
}
}

4
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/HandlerMethodArgumentResolverComposite.java
View File

@@ -37,10 +37,10 @@ import org.springframework.util.Assert;
*/
public class HandlerMethodArgumentResolverComposite implements HandlerMethodArgumentResolver {
private final List<HandlerMethodArgumentResolver> argumentResolvers = new LinkedList<HandlerMethodArgumentResolver>();
private final List<HandlerMethodArgumentResolver> argumentResolvers = new LinkedList<>();
private final Map<MethodParameter, HandlerMethodArgumentResolver> argumentResolverCache =
new ConcurrentHashMap<MethodParameter, HandlerMethodArgumentResolver>(256);
new ConcurrentHashMap<>(256);
/**

4
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/HandlerMethodReturnValueHandlerComposite.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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.
@@ -38,7 +38,7 @@ public class HandlerMethodReturnValueHandlerComposite implements AsyncHandlerMet
private static final Log logger = LogFactory.getLog(HandlerMethodReturnValueHandlerComposite.class);
private final List<HandlerMethodReturnValueHandler> returnValueHandlers = new ArrayList<HandlerMethodReturnValueHandler>();
private final List<HandlerMethodReturnValueHandler> returnValueHandlers = new ArrayList<>();
/**

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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.
@@ -30,7 +30,7 @@ import org.springframework.messaging.Message;
public abstract class SimpAttributesContextHolder {
private static final ThreadLocal<SimpAttributes> attributesHolder =
new NamedThreadLocal<SimpAttributes>("SiMP session attributes");
new NamedThreadLocal<>("SiMP session attributes");
/**

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

@@ -129,7 +129,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
this.clientMessagingTemplate = new SimpMessagingTemplate(clientOutboundChannel);
this.brokerTemplate = brokerTemplate;
Collection<MessageConverter> converters = new ArrayList<MessageConverter>();
Collection<MessageConverter> converters = new ArrayList<>();
converters.add(new StringMessageConverter());
converters.add(new ByteArrayMessageConverter());
this.messageConverter = new CompositeMessageConverter(converters);
@@ -154,7 +154,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
if (CollectionUtils.isEmpty(prefixes)) {
return prefixes;
}
Collection<String> result = new ArrayList<String>(prefixes.size());
Collection<String> result = new ArrayList<>(prefixes.size());
for (String prefix : prefixes) {
if (!prefix.endsWith("/")) {
prefix = prefix + "/";
@@ -303,7 +303,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
ConfigurableBeanFactory beanFactory = (getApplicationContext() instanceof ConfigurableApplicationContext ?
((ConfigurableApplicationContext) getApplicationContext()).getBeanFactory() : null);
List<HandlerMethodArgumentResolver> resolvers = new ArrayList<HandlerMethodArgumentResolver>();
List<HandlerMethodArgumentResolver> resolvers = new ArrayList<>();
// Annotation-based argument resolution
resolvers.add(new HeaderMethodArgumentResolver(this.conversionService, beanFactory));
@@ -322,7 +322,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
@Override
protected List<? extends HandlerMethodReturnValueHandler> initReturnValueHandlers() {
List<HandlerMethodReturnValueHandler> handlers = new ArrayList<HandlerMethodReturnValueHandler>();
List<HandlerMethodReturnValueHandler> handlers = new ArrayList<>();
// Single-purpose return value types
handlers.add(new ListenableFutureReturnValueHandler());
@@ -419,7 +419,7 @@ public class SimpAnnotationMethodMessageHandler extends AbstractMethodMessageHan
@Override
protected Set<String> getDirectLookupDestinations(SimpMessageMappingInfo mapping) {
Set<String> result = new LinkedHashSet<String>();
Set<String> result = new LinkedHashSet<>();
for (String pattern : mapping.getDestinationConditions().getPatterns()) {
if (!this.pathMatcher.isPattern(pattern)) {
result.add(pattern);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -192,7 +192,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
return allMatches;
}
EvaluationContext context = null;
MultiValueMap<String, String> result = new LinkedMultiValueMap<String, String>(allMatches.size());
MultiValueMap<String, String> result = new LinkedMultiValueMap<>(allMatches.size());
for (String sessionId : allMatches.keySet()) {
for (String subId : allMatches.get(sessionId)) {
SessionSubscriptionInfo info = this.subscriptionRegistry.getSubscriptions(sessionId);
@@ -244,7 +244,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
/** Map from destination -> <sessionId, subscriptionId> for fast look-ups */
private final Map<String, LinkedMultiValueMap<String, String>> accessCache =
new ConcurrentHashMap<String, LinkedMultiValueMap<String, String>>(DEFAULT_CACHE_LIMIT);
new ConcurrentHashMap<>(DEFAULT_CACHE_LIMIT);
/** Map from destination -> <sessionId, subscriptionId> with locking */
@SuppressWarnings("serial")
@@ -267,7 +267,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
LinkedMultiValueMap<String, String> result = this.accessCache.get(destination);
if (result == null) {
synchronized (this.updateCache) {
result = new LinkedMultiValueMap<String, String>();
result = new LinkedMultiValueMap<>();
for (SessionSubscriptionInfo info : subscriptionRegistry.getAllSubscriptions()) {
for (String destinationPattern : info.getDestinations()) {
if (getPathMatcher().match(destinationPattern, destination)) {
@@ -301,7 +301,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
public void updateAfterRemovedSubscription(String sessionId, String subsId) {
synchronized (this.updateCache) {
Set<String> destinationsToRemove = new HashSet<String>();
Set<String> destinationsToRemove = new HashSet<>();
for (Map.Entry<String, LinkedMultiValueMap<String, String>> entry : this.updateCache.entrySet()) {
String destination = entry.getKey();
LinkedMultiValueMap<String, String> sessionMap = entry.getValue();
@@ -328,7 +328,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
public void updateAfterRemovedSession(SessionSubscriptionInfo info) {
synchronized (this.updateCache) {
Set<String> destinationsToRemove = new HashSet<String>();
Set<String> destinationsToRemove = new HashSet<>();
for (Map.Entry<String, LinkedMultiValueMap<String, String>> entry : this.updateCache.entrySet()) {
String destination = entry.getKey();
LinkedMultiValueMap<String, String> sessionMap = entry.getValue();
@@ -362,7 +362,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
// sessionId -> SessionSubscriptionInfo
private final ConcurrentMap<String, SessionSubscriptionInfo> sessions =
new ConcurrentHashMap<String, SessionSubscriptionInfo>();
new ConcurrentHashMap<>();
public SessionSubscriptionInfo getSubscriptions(String sessionId) {
return this.sessions.get(sessionId);
@@ -407,7 +407,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
// destination -> subscriptions
private final Map<String, Set<Subscription>> destinationLookup =
new ConcurrentHashMap<String, Set<Subscription>>(4);
new ConcurrentHashMap<>(4);
public SessionSubscriptionInfo(String sessionId) {
Assert.notNull(sessionId, "sessionId must not be null");
@@ -446,7 +446,7 @@ public class DefaultSubscriptionRegistry extends AbstractSubscriptionRegistry {
synchronized (this.destinationLookup) {
subs = this.destinationLookup.get(destination);
if (subs == null) {
subs = new CopyOnWriteArraySet<Subscription>();
subs = new CopyOnWriteArraySet<>();
this.destinationLookup.put(destination, subs);
}
}

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

@@ -48,7 +48,7 @@ public class SimpleBrokerMessageHandler extends AbstractBrokerMessageHandler {
private static final byte[] EMPTY_PAYLOAD = new byte[0];
private final Map<String, SessionInfo> sessions = new ConcurrentHashMap<String, SessionInfo>();
private final Map<String, SessionInfo> sessions = new ConcurrentHashMap<>();
private SubscriptionRegistry subscriptionRegistry;

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

@@ -245,11 +245,11 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
handler.setMessageConverter(brokerMessageConverter());
handler.setValidator(simpValidator());
List<HandlerMethodArgumentResolver> argumentResolvers = new ArrayList<HandlerMethodArgumentResolver>();
List<HandlerMethodArgumentResolver> argumentResolvers = new ArrayList<>();
addArgumentResolvers(argumentResolvers);
handler.setCustomArgumentResolvers(argumentResolvers);
List<HandlerMethodReturnValueHandler> returnValueHandlers = new ArrayList<HandlerMethodReturnValueHandler>();
List<HandlerMethodReturnValueHandler> returnValueHandlers = new ArrayList<>();
addReturnValueHandlers(returnValueHandlers);
handler.setCustomReturnValueHandlers(returnValueHandlers);
@@ -289,7 +289,7 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
if (handler == null) {
return new NoOpBrokerMessageHandler();
}
Map<String, MessageHandler> subscriptions = new HashMap<String, MessageHandler>(1);
Map<String, MessageHandler> subscriptions = new HashMap<>(1);
String destination = getBrokerRegistry().getUserDestinationBroadcast();
if (destination != null) {
subscriptions.put(destination, userDestinationMessageHandler());
@@ -347,7 +347,7 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
@Bean
public CompositeMessageConverter brokerMessageConverter() {
List<MessageConverter> converters = new ArrayList<MessageConverter>();
List<MessageConverter> converters = new ArrayList<>();
boolean registerDefaults = configureMessageConverters(converters);
if (registerDefaults) {
converters.add(new StringMessageConverter());

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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.
@@ -34,7 +34,7 @@ public class ChannelRegistration {
private TaskExecutorRegistration registration;
private final List<ChannelInterceptor> interceptors = new ArrayList<ChannelInterceptor>();
private final List<ChannelInterceptor> interceptors = new ArrayList<>();
/**

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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.
@@ -52,7 +52,7 @@ public class BufferingStompDecoder {
private final int bufferSizeLimit;
private final Queue<ByteBuffer> chunks = new LinkedBlockingQueue<ByteBuffer>();
private final Queue<ByteBuffer> chunks = new LinkedBlockingQueue<>();
private volatile Integer expectedContentLength;
@@ -129,7 +129,7 @@ public class BufferingStompDecoder {
ByteBuffer bufferToDecode = assembleChunksAndReset();
MultiValueMap<String, String> headers = new LinkedMultiValueMap<String, String>();
MultiValueMap<String, String> headers = new LinkedMultiValueMap<>();
List<Message<byte[]>> messages = this.stompDecoder.decode(bufferToDecode, headers);
if (bufferToDecode.hasRemaining()) {

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

@@ -79,7 +79,7 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
private final StompHeaders connectHeaders;
private final SettableListenableFuture<StompSession> sessionFuture = new SettableListenableFuture<StompSession>();
private final SettableListenableFuture<StompSession> sessionFuture = new SettableListenableFuture<>();
private MessageConverter converter = new SimpleMessageConverter();
@@ -96,11 +96,11 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
private final AtomicInteger subscriptionIndex = new AtomicInteger();
private final Map<String, DefaultSubscription> subscriptions = new ConcurrentHashMap<String, DefaultSubscription>(4);
private final Map<String, DefaultSubscription> subscriptions = new ConcurrentHashMap<>(4);
private final AtomicInteger receiptIndex = new AtomicInteger();
private final Map<String, ReceiptHandler> receiptHandlers = new ConcurrentHashMap<String, ReceiptHandler>(4);
private final Map<String, ReceiptHandler> receiptHandlers = new ConcurrentHashMap<>(4);
/* Whether the client is willfully closing the connection */
private volatile boolean closing = false;
@@ -503,9 +503,9 @@ public class DefaultStompSession implements ConnectionHandlingStompSession {
private final String receiptId;
private final List<Runnable> receiptCallbacks = new ArrayList<Runnable>(2);
private final List<Runnable> receiptCallbacks = new ArrayList<>(2);
private final List<Runnable> receiptLostCallbacks = new ArrayList<Runnable>(2);
private final List<Runnable> receiptLostCallbacks = new ArrayList<>(2);
private ScheduledFuture<?> future;

4
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2TcpStompClient.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -61,7 +61,7 @@ public class Reactor2TcpStompClient extends StompClientSupport {
ConfigurationReader reader = new StompClientDispatcherConfigReader();
Environment environment = new Environment(reader).assignErrorJournal();
StompTcpClientSpecFactory factory = new StompTcpClientSpecFactory(environment, host, port);
this.tcpClient = new Reactor2TcpClient<byte[]>(factory);
this.tcpClient = new Reactor2TcpClient<>(factory);
}
/**

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -81,7 +81,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
private static final byte[] EMPTY_PAYLOAD = new byte[0];
private static final ListenableFutureTask<Void> EMPTY_TASK = new ListenableFutureTask<Void>(new VoidCallable());
private static final ListenableFutureTask<Void> EMPTY_TASK = new ListenableFutureTask<>(new VoidCallable());
// STOMP recommends error of margin for receiving heartbeats
private static final long HEARTBEAT_MULTIPLIER = 3;
@@ -122,14 +122,14 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
private String virtualHost;
private final Map<String, MessageHandler> systemSubscriptions = new HashMap<String, MessageHandler>(4);
private final Map<String, MessageHandler> systemSubscriptions = new HashMap<>(4);
private TcpOperations<byte[]> tcpClient;
private MessageHeaderInitializer headerInitializer;
private final Map<String, StompConnectionHandler> connectionHandlers =
new ConcurrentHashMap<String, StompConnectionHandler>();
new ConcurrentHashMap<>();
private final Stats stats = new Stats();
@@ -974,7 +974,7 @@ public class StompBrokerRelayMessageHandler extends AbstractBrokerMessageHandler
private static class StompTcpClientFactory {
public TcpOperations<byte[]> create(String relayHost, int relayPort, Reactor2StompCodec codec) {
return new Reactor2TcpClient<byte[]>(relayHost, relayPort, codec);
return new Reactor2TcpClient<>(relayHost, relayPort, codec);
}
}

4
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompCommand.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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.
@@ -51,7 +51,7 @@ public enum StompCommand {
ERROR;
private static Map<StompCommand, SimpMessageType> messageTypes = new HashMap<StompCommand, SimpMessageType>();
private static Map<StompCommand, SimpMessageType> messageTypes = new HashMap<>();
static {
messageTypes.put(StompCommand.CONNECT, SimpMessageType.CONNECT);
messageTypes.put(StompCommand.STOMP, SimpMessageType.CONNECT);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -106,7 +106,7 @@ public class StompDecoder {
* @throws StompConversionException raised in case of decoding issues
*/
public List<Message<byte[]>> decode(ByteBuffer buffer, MultiValueMap<String, String> partialMessageHeaders) {
List<Message<byte[]>> messages = new ArrayList<Message<byte[]>>();
List<Message<byte[]>> messages = new ArrayList<>();
while (buffer.hasRemaining()) {
Message<byte[]> message = decodeMessage(buffer, partialMessageHeaders);
if (message != null) {

12
spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaders.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -105,13 +105,13 @@ public class StompHeaders implements MultiValueMap<String, String>, Serializable
* Create a new instance to be populated with new header values.
*/
public StompHeaders() {
this(new LinkedMultiValueMap<String, String>(4), false);
this(new LinkedMultiValueMap<>(4), false);
}
private StompHeaders(Map<String, List<String>> headers, boolean readOnly) {
Assert.notNull(headers, "'headers' must not be null");
if (readOnly) {
Map<String, List<String>> map = new LinkedMultiValueMap<String, String>(headers.size());
Map<String, List<String>> map = new LinkedMultiValueMap<>(headers.size());
for (Entry<String, List<String>> entry : headers.entrySet()) {
List<String> values = Collections.unmodifiableList(entry.getValue());
map.put(entry.getKey(), values);
@@ -394,7 +394,7 @@ public class StompHeaders implements MultiValueMap<String, String>, Serializable
public void add(String headerName, String headerValue) {
List<String> headerValues = headers.get(headerName);
if (headerValues == null) {
headerValues = new LinkedList<String>();
headerValues = new LinkedList<>();
this.headers.put(headerName, headerValues);
}
headerValues.add(headerValue);
@@ -410,7 +410,7 @@ public class StompHeaders implements MultiValueMap<String, String>, Serializable
*/
@Override
public void set(String headerName, String headerValue) {
List<String> headerValues = new LinkedList<String>();
List<String> headerValues = new LinkedList<>();
headerValues.add(headerValue);
headers.put(headerName, headerValues);
}
@@ -424,7 +424,7 @@ public class StompHeaders implements MultiValueMap<String, String>, Serializable
@Override
public Map<String, String> toSingleValueMap() {
LinkedHashMap<String, String> singleValueMap = new LinkedHashMap<String,String>(this.headers.size());
LinkedHashMap<String, String> singleValueMap = new LinkedHashMap<>(this.headers.size());
for (Entry<String, List<String>> entry : headers.entrySet()) {
singleValueMap.put(entry.getKey(), entry.getValue().get(0));
}

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

@@ -126,7 +126,7 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver {
return null;
}
String user = parseResult.getUser();
Set<String> targetSet = new HashSet<String>();
Set<String> targetSet = new HashSet<>();
for (String sessionId : parseResult.getSessionIds()) {
String actualDestination = parseResult.getActualDestination();
String targetDestination = getTargetDestination(sourceDestination, actualDestination, sessionId, user);
@@ -181,7 +181,7 @@ public class DefaultUserDestinationResolver implements UserDestinationResolver {
}
else {
Set<SimpSession> sessions = user.getSessions();
sessionIds = new HashSet<String>(sessions.size());
sessionIds = new HashSet<>(sessions.size());
for (SimpSession session : sessions) {
sessionIds.add(session.getId());
}

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

@@ -51,7 +51,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
private final SimpUserRegistry localRegistry;
private final Map<String, UserRegistrySnapshot> remoteRegistries = new ConcurrentHashMap<String, UserRegistrySnapshot>();
private final Map<String, UserRegistrySnapshot> remoteRegistries = new ConcurrentHashMap<>();
private final boolean delegateApplicationEvents;
@@ -125,7 +125,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
@Override
public Set<SimpUser> getUsers() {
// Prefer remote registries due to cross-server SessionLookup
Set<SimpUser> result = new HashSet<SimpUser>();
Set<SimpUser> result = new HashSet<>();
for (UserRegistrySnapshot registry : this.remoteRegistries.values()) {
result.addAll(registry.getUserMap().values());
}
@@ -135,7 +135,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
@Override
public Set<SimpSubscription> findSubscriptions(SimpSubscriptionMatcher matcher) {
Set<SimpSubscription> result = new HashSet<SimpSubscription>();
Set<SimpSubscription> result = new HashSet<>();
for (UserRegistrySnapshot registry : this.remoteRegistries.values()) {
result.addAll(registry.findSubscriptions(matcher));
}
@@ -201,7 +201,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
public UserRegistrySnapshot(String id, SimpUserRegistry registry) {
this.id = id;
Set<SimpUser> users = registry.getUsers();
this.users = new HashMap<String, TransferSimpUser>(users.size());
this.users = new HashMap<>(users.size());
for (SimpUser user : users) {
this.users.put(user.getName(), new TransferSimpUser(user));
}
@@ -238,7 +238,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
public Set<SimpSubscription> findSubscriptions(SimpSubscriptionMatcher matcher) {
Set<SimpSubscription> result = new HashSet<SimpSubscription>();
Set<SimpSubscription> result = new HashSet<>();
for (TransferSimpUser user : this.users.values()) {
for (TransferSimpSession session : user.sessions) {
for (SimpSubscription subscription : session.subscriptions) {
@@ -279,7 +279,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
* Default constructor for JSON deserialization.
*/
public TransferSimpUser() {
this.sessions = new HashSet<TransferSimpSession>(1);
this.sessions = new HashSet<>(1);
}
/**
@@ -288,7 +288,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
public TransferSimpUser(SimpUser user) {
this.name = user.getName();
Set<SimpSession> sessions = user.getSessions();
this.sessions = new HashSet<TransferSimpSession>(sessions.size());
this.sessions = new HashSet<>(sessions.size());
for (SimpSession session : sessions) {
this.sessions.add(new TransferSimpSession(session));
}
@@ -333,9 +333,9 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
public Set<SimpSession> getSessions() {
if (this.sessionLookup != null) {
Map<String, SimpSession> sessions = this.sessionLookup.findSessions(getName());
return new HashSet<SimpSession>(sessions.values());
return new HashSet<>(sessions.values());
}
return new HashSet<SimpSession>(this.sessions);
return new HashSet<>(this.sessions);
}
private void afterDeserialization(SessionLookup sessionLookup) {
@@ -386,7 +386,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
* Default constructor for JSON deserialization.
*/
public TransferSimpSession() {
this.subscriptions = new HashSet<TransferSimpSubscription>(4);
this.subscriptions = new HashSet<>(4);
}
/**
@@ -395,7 +395,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
public TransferSimpSession(SimpSession session) {
this.id = session.getId();
Set<SimpSubscription> subscriptions = session.getSubscriptions();
this.subscriptions = new HashSet<TransferSimpSubscription>(subscriptions.size());
this.subscriptions = new HashSet<>(subscriptions.size());
for (SimpSubscription subscription : subscriptions) {
this.subscriptions.add(new TransferSimpSubscription(subscription));
}
@@ -425,7 +425,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
@Override
public Set<SimpSubscription> getSubscriptions() {
return new HashSet<SimpSubscription>(this.subscriptions);
return new HashSet<>(this.subscriptions);
}
private void afterDeserialization() {
@@ -536,7 +536,7 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
private class SessionLookup {
public Map<String, SimpSession> findSessions(String userName) {
Map<String, SimpSession> map = new HashMap<String, SimpSession>(1);
Map<String, SimpSession> map = new HashMap<>(1);
SimpUser user = localRegistry.getUser(userName);
if (user != null) {
for (SimpSession session : user.getSessions()) {

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

@@ -41,7 +41,7 @@ public abstract class AbstractMessageChannel implements MessageChannel, Intercep
protected final Log logger = LogFactory.getLog(getClass());
private final List<ChannelInterceptor> interceptors = new ArrayList<ChannelInterceptor>(5);
private final List<ChannelInterceptor> interceptors = new ArrayList<>(5);
private String beanName;

4
spring-messaging/src/main/java/org/springframework/messaging/support/AbstractSubscribableChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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.
@@ -31,7 +31,7 @@ import org.springframework.messaging.SubscribableChannel;
*/
public abstract class AbstractSubscribableChannel extends AbstractMessageChannel implements SubscribableChannel {
private final Set<MessageHandler> handlers = new CopyOnWriteArraySet<MessageHandler>();
private final Set<MessageHandler> handlers = new CopyOnWriteArraySet<>();
public Set<MessageHandler> getSubscribers() {

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

@@ -37,7 +37,7 @@ public class ExecutorSubscribableChannel extends AbstractSubscribableChannel {
private final Executor executor;
private final List<ExecutorChannelInterceptor> executorInterceptors = new ArrayList<ExecutorChannelInterceptor>(4);
private final List<ExecutorChannelInterceptor> executorInterceptors = new ArrayList<>(4);
/**

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 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.
@@ -153,7 +153,7 @@ public final class MessageBuilder<T> {
return (Message<T>) new ErrorMessage((Throwable) this.payload, headersToUse);
}
else {
return new GenericMessage<T>(this.payload, headersToUse);
return new GenericMessage<>(this.payload, headersToUse);
}
}
@@ -165,7 +165,7 @@ public final class MessageBuilder<T> {
* @param message the Message from which the payload and all headers will be copied
*/
public static <T> MessageBuilder<T> fromMessage(Message<T> message) {
return new MessageBuilder<T>(message);
return new MessageBuilder<>(message);
}
/**
@@ -173,7 +173,7 @@ public final class MessageBuilder<T> {
* @param payload the payload
*/
public static <T> MessageBuilder<T> withPayload(T payload) {
return new MessageBuilder<T>(payload, new MessageHeaderAccessor());
return new MessageBuilder<>(payload, new MessageHeaderAccessor());
}
/**
@@ -194,7 +194,7 @@ public final class MessageBuilder<T> {
return (Message<T>) new ErrorMessage((Throwable) payload, messageHeaders);
}
else {
return new GenericMessage<T>(payload, messageHeaders);
return new GenericMessage<>(payload, messageHeaders);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -282,7 +282,7 @@ public class MessageHeaderAccessor {
* where each new call returns a fresh copy of the current header values.
*/
public Map<String, Object> toMap() {
return new HashMap<String, Object>(this.headers);
return new HashMap<>(this.headers);
}
@@ -354,7 +354,7 @@ public class MessageHeaderAccessor {
* names. Supported pattern styles are: "xxx*", "*xxx", "*xxx*" and "xxx*yyy".
*/
public void removeHeaders(String... headerPatterns) {
List<String> headersToRemove = new ArrayList<String>();
List<String> headersToRemove = new ArrayList<>();
for (String pattern : headerPatterns) {
if (StringUtils.hasLength(pattern)){
if (pattern.contains("*")){
@@ -371,7 +371,7 @@ public class MessageHeaderAccessor {
}
private List<String> getMatchingHeaderNames(String pattern, Map<String, Object> headers) {
List<String> matchingHeaderNames = new ArrayList<String>();
List<String> matchingHeaderNames = new ArrayList<>();
if (headers != null) {
for (String key : headers.keySet()) {
if (PatternMatchUtils.simpleMatch(pattern, key)) {

16
spring-messaging/src/main/java/org/springframework/messaging/support/NativeMessageHeaderAccessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -64,7 +64,7 @@ public class NativeMessageHeaderAccessor extends MessageHeaderAccessor {
*/
protected NativeMessageHeaderAccessor(Map<String, List<String>> nativeHeaders) {
if (!CollectionUtils.isEmpty(nativeHeaders)) {
setHeader(NATIVE_HEADERS, new LinkedMultiValueMap<String, String>(nativeHeaders));
setHeader(NATIVE_HEADERS, new LinkedMultiValueMap<>(nativeHeaders));
}
}
@@ -79,7 +79,7 @@ public class NativeMessageHeaderAccessor extends MessageHeaderAccessor {
if (map != null) {
// Force removal since setHeader checks for equality
removeHeader(NATIVE_HEADERS);
setHeader(NATIVE_HEADERS, new LinkedMultiValueMap<String, String>(map));
setHeader(NATIVE_HEADERS, new LinkedMultiValueMap<>(map));
}
}
}
@@ -94,7 +94,7 @@ public class NativeMessageHeaderAccessor extends MessageHeaderAccessor {
*/
public Map<String, List<String>> toNativeHeaderMap() {
Map<String, List<String>> map = getNativeHeaders();
return (map != null ? new LinkedMultiValueMap<String, String>(map) : Collections.<String, List<String>>emptyMap());
return (map != null ? new LinkedMultiValueMap<>(map) : Collections.<String, List<String>>emptyMap());
}
@Override
@@ -154,10 +154,10 @@ public class NativeMessageHeaderAccessor extends MessageHeaderAccessor {
return;
}
if (map == null) {
map = new LinkedMultiValueMap<String, String>(4);
map = new LinkedMultiValueMap<>(4);
setHeader(NATIVE_HEADERS, map);
}
List<String> values = new LinkedList<String>();
List<String> values = new LinkedList<>();
values.add(value);
if (!ObjectUtils.nullSafeEquals(values, getHeader(name))) {
setModified(true);
@@ -175,12 +175,12 @@ public class NativeMessageHeaderAccessor extends MessageHeaderAccessor {
}
Map<String, List<String>> nativeHeaders = getNativeHeaders();
if (nativeHeaders == null) {
nativeHeaders = new LinkedMultiValueMap<String, String>(4);
nativeHeaders = new LinkedMultiValueMap<>(4);
setHeader(NATIVE_HEADERS, nativeHeaders);
}
List<String> values = nativeHeaders.get(name);
if (values == null) {
values = new LinkedList<String>();
values = new LinkedList<>();
nativeHeaders.put(name, values);
}
values.add(value);

2
spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/AbstractPromiseToListenableFutureAdapter.java
View File

@@ -43,7 +43,7 @@ abstract class AbstractPromiseToListenableFutureAdapter<S, T> implements Listena
private final Promise<S> promise;
private final ListenableFutureCallbackRegistry<T> registry = new ListenableFutureCallbackRegistry<T>();
private final ListenableFutureCallbackRegistry<T> registry = new ListenableFutureCallbackRegistry<>();
protected AbstractPromiseToListenableFutureAdapter(Promise<S> promise) {

20
spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpClient.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -89,7 +89,7 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
private final TcpClientFactory<Message<P>, Message<P>> tcpClientSpecFactory;
private final List<TcpClient<Message<P>, Message<P>>> tcpClients =
new ArrayList<TcpClient<Message<P>, Message<P>>>();
new ArrayList<>();
private boolean stopping;
@@ -173,7 +173,7 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
if (this.stopping) {
IllegalStateException ex = new IllegalStateException("Shutting down.");
connectionHandler.afterConnectFailure(ex);
return new PassThroughPromiseToListenableFutureAdapter<Void>(Promises.<Void>error(ex));
return new PassThroughPromiseToListenableFutureAdapter<>(Promises.<Void>error(ex));
}
tcpClient = NetStreams.tcpClient(REACTOR_TCP_CLIENT_TYPE, this.tcpClientSpecFactory);
this.tcpClients.add(tcpClient);
@@ -188,9 +188,9 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
}
Promise<Void> promise = tcpClient.start(
new MessageChannelStreamHandler<P>(connectionHandler, cleanupTask));
new MessageChannelStreamHandler<>(connectionHandler, cleanupTask));
return new PassThroughPromiseToListenableFutureAdapter<Void>(
return new PassThroughPromiseToListenableFutureAdapter<>(
promise.onError(new Consumer<Throwable>() {
@Override
public void accept(Throwable ex) {
@@ -212,7 +212,7 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
if (this.stopping) {
IllegalStateException ex = new IllegalStateException("Shutting down.");
connectionHandler.afterConnectFailure(ex);
return new PassThroughPromiseToListenableFutureAdapter<Void>(Promises.<Void>error(ex));
return new PassThroughPromiseToListenableFutureAdapter<>(Promises.<Void>error(ex));
}
tcpClient = NetStreams.tcpClient(REACTOR_TCP_CLIENT_TYPE, this.tcpClientSpecFactory);
this.tcpClients.add(tcpClient);
@@ -227,10 +227,10 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
}
Stream<Tuple2<InetSocketAddress, Integer>> stream = tcpClient.start(
new MessageChannelStreamHandler<P>(connectionHandler, cleanupTask),
new MessageChannelStreamHandler<>(connectionHandler, cleanupTask),
new ReactorReconnectAdapter(strategy));
return new PassThroughPromiseToListenableFutureAdapter<Void>(stream.next().after());
return new PassThroughPromiseToListenableFutureAdapter<>(stream.next().after());
}
@Override
@@ -283,7 +283,7 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
});
}
return new PassThroughPromiseToListenableFutureAdapter<Void>(promise);
return new PassThroughPromiseToListenableFutureAdapter<>(promise);
}
@@ -322,7 +322,7 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
@Override
public Publisher<Void> apply(ChannelStream<Message<P>, Message<P>> channelStream) {
Promise<Void> closePromise = Promises.prepare();
this.connectionHandler.afterConnected(new Reactor2TcpConnection<P>(channelStream, closePromise));
this.connectionHandler.afterConnected(new Reactor2TcpConnection<>(channelStream, closePromise));
channelStream
.finallyDo(new Consumer<Signal<Message<P>>>() {
@Override

4
spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpConnection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -53,7 +53,7 @@ public class Reactor2TcpConnection<P> implements TcpConnection<P> {
public ListenableFuture<Void> send(Message<P> message) {
Promise<Void> afterWrite = Promises.prepare();
this.channelStream.writeWith(Streams.just(message)).subscribe(afterWrite);
return new PassThroughPromiseToListenableFutureAdapter<Void>(afterWrite);
return new PassThroughPromiseToListenableFutureAdapter<>(afterWrite);
}
@Override