Polishing
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2017 the original author or authors.
|
||||
* Copyright 2002-2019 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -38,7 +38,7 @@ import org.springframework.web.socket.messaging.SubProtocolWebSocketHandler;
|
||||
* {@code @EnableWebSocketMessageBroker} for Java config and
|
||||
* {@code <websocket:message-broker>} for XML.
|
||||
*
|
||||
* <p>By default aggregated information is logged every 15 minutes at INFO level.
|
||||
* <p>By default aggregated information is logged every 30 minutes at INFO level.
|
||||
* The frequency of logging can be changed via {@link #setLoggingPeriod(long)}.
|
||||
*
|
||||
* <p>This class is declared as a Spring bean by the above configuration with the
|
||||
@@ -189,7 +189,7 @@ public class WebSocketMessageBrokerStats {
|
||||
", stompSubProtocol[" + getStompSubProtocolStatsInfo() + "]" +
|
||||
", stompBrokerRelay[" + getStompBrokerRelayStatsInfo() + "]" +
|
||||
", inboundChannel[" + getClientInboundExecutorStatsInfo() + "]" +
|
||||
", outboundChannel" + getClientOutboundExecutorStatsInfo() + "]" +
|
||||
", outboundChannel[" + getClientOutboundExecutorStatsInfo() + "]" +
|
||||
", sockJsScheduler[" + getSockJsTaskSchedulerStatsInfo() + "]";
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2018 the original author or authors.
|
||||
* Copyright 2002-2019 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -22,6 +22,7 @@ import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ScheduledThreadPoolExecutor;
|
||||
|
||||
import org.hamcrest.Matchers;
|
||||
import org.junit.Test;
|
||||
@@ -92,8 +93,8 @@ import static org.hamcrest.Matchers.*;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
* Test fixture for MessageBrokerBeanDefinitionParser.
|
||||
* See test configuration files websocket-config-broker-*.xml.
|
||||
* Test fixture for {@link MessageBrokerBeanDefinitionParser}.
|
||||
* Also see test configuration files websocket-config-broker-*.xml.
|
||||
*
|
||||
* @author Brian Clozel
|
||||
* @author Artem Bilan
|
||||
@@ -174,8 +175,9 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
assertFalse(defaultSockJsService.shouldSuppressCors());
|
||||
|
||||
ThreadPoolTaskScheduler scheduler = (ThreadPoolTaskScheduler) defaultSockJsService.getTaskScheduler();
|
||||
assertEquals(Runtime.getRuntime().availableProcessors(), scheduler.getScheduledThreadPoolExecutor().getCorePoolSize());
|
||||
assertTrue(scheduler.getScheduledThreadPoolExecutor().getRemoveOnCancelPolicy());
|
||||
ScheduledThreadPoolExecutor executor = scheduler.getScheduledThreadPoolExecutor();
|
||||
assertEquals(Runtime.getRuntime().availableProcessors(), executor.getCorePoolSize());
|
||||
assertTrue(executor.getRemoveOnCancelPolicy());
|
||||
|
||||
interceptors = defaultSockJsService.getHandshakeInterceptors();
|
||||
assertThat(interceptors, contains(instanceOf(FooTestInterceptor.class),
|
||||
@@ -199,14 +201,13 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
SimpleBrokerMessageHandler brokerMessageHandler = this.appContext.getBean(SimpleBrokerMessageHandler.class);
|
||||
assertNotNull(brokerMessageHandler);
|
||||
Collection<String> prefixes = brokerMessageHandler.getDestinationPrefixes();
|
||||
assertEquals(Arrays.asList("/topic", "/queue"), new ArrayList<String>(prefixes));
|
||||
assertEquals(Arrays.asList("/topic", "/queue"), new ArrayList<>(prefixes));
|
||||
DefaultSubscriptionRegistry registry = (DefaultSubscriptionRegistry) brokerMessageHandler.getSubscriptionRegistry();
|
||||
assertEquals("my-selector", registry.getSelectorHeaderName());
|
||||
assertNotNull(brokerMessageHandler.getTaskScheduler());
|
||||
assertArrayEquals(new long[] {15000, 15000}, brokerMessageHandler.getHeartbeatValue());
|
||||
|
||||
List<Class<? extends MessageHandler>> subscriberTypes =
|
||||
Arrays.<Class<? extends MessageHandler>>asList(SimpAnnotationMethodMessageHandler.class,
|
||||
List<Class<? extends MessageHandler>> subscriberTypes = Arrays.asList(SimpAnnotationMethodMessageHandler.class,
|
||||
UserDestinationMessageHandler.class, SimpleBrokerMessageHandler.class);
|
||||
testChannel("clientInboundChannel", subscriberTypes, 2);
|
||||
testExecutor("clientInboundChannel", Runtime.getRuntime().availableProcessors() * 2, Integer.MAX_VALUE, 60);
|
||||
@@ -215,8 +216,7 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
testChannel("clientOutboundChannel", subscriberTypes, 1);
|
||||
testExecutor("clientOutboundChannel", Runtime.getRuntime().availableProcessors() * 2, Integer.MAX_VALUE, 60);
|
||||
|
||||
subscriberTypes = Arrays.<Class<? extends MessageHandler>>asList(
|
||||
SimpleBrokerMessageHandler.class, UserDestinationMessageHandler.class);
|
||||
subscriberTypes = Arrays.asList(SimpleBrokerMessageHandler.class, UserDestinationMessageHandler.class);
|
||||
testChannel("brokerChannel", subscriberTypes, 1);
|
||||
try {
|
||||
this.appContext.getBean("brokerChannelExecutor", ThreadPoolTaskExecutor.class);
|
||||
@@ -228,8 +228,9 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
|
||||
assertNotNull(this.appContext.getBean("webSocketScopeConfigurer", CustomScopeConfigurer.class));
|
||||
|
||||
DirectFieldAccessor subscriptionRegistryAccessor = new DirectFieldAccessor(brokerMessageHandler.getSubscriptionRegistry());
|
||||
String pathSeparator = (String) new DirectFieldAccessor(subscriptionRegistryAccessor.getPropertyValue("pathMatcher")).getPropertyValue("pathSeparator");
|
||||
DirectFieldAccessor accessor = new DirectFieldAccessor(registry);
|
||||
Object pathMatcher = accessor.getPropertyValue("pathMatcher");
|
||||
String pathSeparator = (String) new DirectFieldAccessor(pathMatcher).getPropertyValue("pathSeparator");
|
||||
assertEquals(".", pathSeparator);
|
||||
}
|
||||
|
||||
@@ -274,9 +275,8 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
assertEquals(5000, messageBroker.getSystemHeartbeatSendInterval());
|
||||
assertThat(messageBroker.getDestinationPrefixes(), Matchers.containsInAnyOrder("/topic","/queue"));
|
||||
|
||||
List<Class<? extends MessageHandler>> subscriberTypes =
|
||||
Arrays.<Class<? extends MessageHandler>>asList(SimpAnnotationMethodMessageHandler.class,
|
||||
UserDestinationMessageHandler.class, StompBrokerRelayMessageHandler.class);
|
||||
List<Class<? extends MessageHandler>> subscriberTypes = Arrays.asList(SimpAnnotationMethodMessageHandler.class,
|
||||
UserDestinationMessageHandler.class, StompBrokerRelayMessageHandler.class);
|
||||
testChannel("clientInboundChannel", subscriberTypes, 2);
|
||||
testExecutor("clientInboundChannel", Runtime.getRuntime().availableProcessors() * 2, Integer.MAX_VALUE, 60);
|
||||
|
||||
@@ -284,8 +284,7 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
testChannel("clientOutboundChannel", subscriberTypes, 1);
|
||||
testExecutor("clientOutboundChannel", Runtime.getRuntime().availableProcessors() * 2, Integer.MAX_VALUE, 60);
|
||||
|
||||
subscriberTypes = Arrays.<Class<? extends MessageHandler>>asList(
|
||||
StompBrokerRelayMessageHandler.class, UserDestinationMessageHandler.class);
|
||||
subscriberTypes = Arrays.asList(StompBrokerRelayMessageHandler.class, UserDestinationMessageHandler.class);
|
||||
testChannel("brokerChannel", subscriberTypes, 1);
|
||||
try {
|
||||
this.appContext.getBean("brokerChannelExecutor", ThreadPoolTaskExecutor.class);
|
||||
@@ -316,10 +315,14 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
String expected = "WebSocketSession\\[0 current WS\\(0\\)-HttpStream\\(0\\)-HttpPoll\\(0\\), " +
|
||||
"0 total, 0 closed abnormally \\(0 connect failure, 0 send limit, 0 transport error\\)\\], " +
|
||||
"stompSubProtocol\\[processed CONNECT\\(0\\)-CONNECTED\\(0\\)-DISCONNECT\\(0\\)\\], " +
|
||||
"stompBrokerRelay\\[0 sessions, relayhost:1234 \\(not available\\), processed CONNECT\\(0\\)-CONNECTED\\(0\\)-DISCONNECT\\(0\\)\\], " +
|
||||
"inboundChannel\\[pool size = \\d, active threads = \\d, queued tasks = \\d, completed tasks = \\d\\], " +
|
||||
"outboundChannelpool size = \\d, active threads = \\d, queued tasks = \\d, completed tasks = \\d\\], " +
|
||||
"sockJsScheduler\\[pool size = \\d, active threads = \\d, queued tasks = \\d, completed tasks = \\d\\]";
|
||||
"stompBrokerRelay\\[0 sessions, relayhost:1234 \\(not available\\), " +
|
||||
"processed CONNECT\\(0\\)-CONNECTED\\(0\\)-DISCONNECT\\(0\\)\\], " +
|
||||
"inboundChannel\\[pool size = \\d, active threads = \\d, queued tasks = \\d, " +
|
||||
"completed tasks = \\d\\], " +
|
||||
"outboundChannel\\[pool size = \\d, active threads = \\d, queued tasks = \\d, " +
|
||||
"completed tasks = \\d\\], " +
|
||||
"sockJsScheduler\\[pool size = \\d, active threads = \\d, queued tasks = \\d, " +
|
||||
"completed tasks = \\d\\]";
|
||||
|
||||
assertTrue("\nExpected: " + expected.replace("\\", "") + "\n Actual: " + actual, actual.matches(expected));
|
||||
}
|
||||
@@ -355,7 +358,8 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
assertEquals(MimeTypeUtils.APPLICATION_JSON, ((DefaultContentTypeResolver) resolver).getDefaultMimeType());
|
||||
|
||||
DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(annotationMethodMessageHandler);
|
||||
String pathSeparator = (String) new DirectFieldAccessor(handlerAccessor.getPropertyValue("pathMatcher")).getPropertyValue("pathSeparator");
|
||||
Object pathMatcher = handlerAccessor.getPropertyValue("pathMatcher");
|
||||
String pathSeparator = (String) new DirectFieldAccessor(pathMatcher).getPropertyValue("pathSeparator");
|
||||
assertEquals(".", pathSeparator);
|
||||
}
|
||||
|
||||
@@ -371,9 +375,8 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
assertSame(this.appContext.getBean("myValidator"), validator);
|
||||
assertThat(validator, Matchers.instanceOf(TestValidator.class));
|
||||
|
||||
List<Class<? extends MessageHandler>> subscriberTypes =
|
||||
Arrays.<Class<? extends MessageHandler>>asList(SimpAnnotationMethodMessageHandler.class,
|
||||
UserDestinationMessageHandler.class, SimpleBrokerMessageHandler.class);
|
||||
List<Class<? extends MessageHandler>> subscriberTypes = Arrays.asList(SimpAnnotationMethodMessageHandler.class,
|
||||
UserDestinationMessageHandler.class, SimpleBrokerMessageHandler.class);
|
||||
|
||||
testChannel("clientInboundChannel", subscriberTypes, 3);
|
||||
testExecutor("clientInboundChannel", 100, 200, 600);
|
||||
@@ -383,16 +386,13 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
testChannel("clientOutboundChannel", subscriberTypes, 3);
|
||||
testExecutor("clientOutboundChannel", 101, 201, 601);
|
||||
|
||||
subscriberTypes = Arrays.<Class<? extends MessageHandler>>asList(SimpleBrokerMessageHandler.class,
|
||||
UserDestinationMessageHandler.class);
|
||||
subscriberTypes = Arrays.asList(SimpleBrokerMessageHandler.class, UserDestinationMessageHandler.class);
|
||||
|
||||
testChannel("brokerChannel", subscriberTypes, 1);
|
||||
testExecutor("brokerChannel", 102, 202, 602);
|
||||
}
|
||||
|
||||
// SPR-11623
|
||||
|
||||
@Test
|
||||
@Test // SPR-11623
|
||||
public void customChannelsWithDefaultExecutor() {
|
||||
loadBeanDefinitions("websocket-config-broker-customchannels-default-executor.xml");
|
||||
|
||||
@@ -441,27 +441,23 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
}
|
||||
|
||||
|
||||
private void testChannel(String channelName, List<Class<? extends MessageHandler>> subscriberTypes,
|
||||
int interceptorCount) {
|
||||
private void testChannel(
|
||||
String channelName, List<Class<? extends MessageHandler>> subscriberTypes, int interceptorCount) {
|
||||
|
||||
AbstractSubscribableChannel channel = this.appContext.getBean(channelName, AbstractSubscribableChannel.class);
|
||||
|
||||
for (Class<? extends MessageHandler> subscriberType : subscriberTypes) {
|
||||
MessageHandler subscriber = this.appContext.getBean(subscriberType);
|
||||
assertNotNull("No subsription for " + subscriberType, subscriber);
|
||||
assertNotNull("No subscription for " + subscriberType, subscriber);
|
||||
assertTrue(channel.hasSubscription(subscriber));
|
||||
}
|
||||
|
||||
List<ChannelInterceptor> interceptors = channel.getInterceptors();
|
||||
assertEquals(interceptorCount, interceptors.size());
|
||||
assertEquals(ImmutableMessageChannelInterceptor.class, interceptors.get(interceptors.size()-1).getClass());
|
||||
}
|
||||
|
||||
private void testExecutor(String channelName, int corePoolSize, int maxPoolSize, int keepAliveSeconds) {
|
||||
|
||||
ThreadPoolTaskExecutor taskExecutor =
|
||||
this.appContext.getBean(channelName + "Executor", ThreadPoolTaskExecutor.class);
|
||||
|
||||
assertEquals(corePoolSize, taskExecutor.getCorePoolSize());
|
||||
assertEquals(maxPoolSize, taskExecutor.getMaxPoolSize());
|
||||
assertEquals(keepAliveSeconds, taskExecutor.getKeepAliveSeconds());
|
||||
@@ -479,6 +475,7 @@ public class MessageBrokerBeanDefinitionParserTests {
|
||||
return (handler instanceof WebSocketHandlerDecorator) ?
|
||||
((WebSocketHandlerDecorator) handler).getLastHandler() : handler;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
@@ -505,7 +502,6 @@ class CustomReturnValueHandler implements HandlerMethodReturnValueHandler {
|
||||
|
||||
@Override
|
||||
public void handleReturnValue(Object returnValue, MethodParameter returnType, Message<?> message) throws Exception {
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@@ -536,12 +532,15 @@ class TestWebSocketHandlerDecorator extends WebSocketHandlerDecorator {
|
||||
class TestStompErrorHandler extends StompSubProtocolErrorHandler {
|
||||
}
|
||||
|
||||
|
||||
class TestValidator implements Validator {
|
||||
|
||||
@Override
|
||||
public boolean supports(Class<?> clazz) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void validate(Object target, Errors errors) { }
|
||||
public void validate(Object target, Errors errors) {
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2015 the original author or authors.
|
||||
* Copyright 2002-2019 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -16,9 +16,6 @@
|
||||
|
||||
package org.springframework.web.socket.config.annotation;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
@@ -51,7 +48,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
import org.springframework.stereotype.Controller;
|
||||
import org.springframework.web.servlet.HandlerMapping;
|
||||
import org.springframework.web.servlet.handler.SimpleUrlHandlerMapping;
|
||||
import org.springframework.web.socket.TextMessage;
|
||||
import org.springframework.web.socket.WebSocketHandler;
|
||||
import org.springframework.web.socket.WebSocketSession;
|
||||
import org.springframework.web.socket.config.WebSocketMessageBrokerStats;
|
||||
@@ -64,9 +60,11 @@ import org.springframework.web.socket.messaging.SubProtocolHandler;
|
||||
import org.springframework.web.socket.messaging.SubProtocolWebSocketHandler;
|
||||
import org.springframework.web.socket.server.support.WebSocketHttpRequestHandler;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/**
|
||||
* Test fixture for
|
||||
* {@link org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurationSupport}.
|
||||
* Test fixture for {@link WebSocketMessageBrokerConfigurationSupport}.
|
||||
*
|
||||
* @author Rossen Stoyanchev
|
||||
*/
|
||||
@@ -96,8 +94,8 @@ public class WebSocketMessageBrokerConfigurationSupportTests {
|
||||
session.setOpen(true);
|
||||
webSocketHandler.afterConnectionEstablished(session);
|
||||
|
||||
TextMessage textMessage = StompTextMessageBuilder.create(StompCommand.SEND).headers("destination:/foo").build();
|
||||
webSocketHandler.handleMessage(session, textMessage);
|
||||
webSocketHandler.handleMessage(session,
|
||||
StompTextMessageBuilder.create(StompCommand.SEND).headers("destination:/foo").build());
|
||||
|
||||
Message<?> message = channel.messages.get(0);
|
||||
StompHeaderAccessor accessor = StompHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
|
||||
@@ -173,7 +171,7 @@ public class WebSocketMessageBrokerConfigurationSupportTests {
|
||||
"stompSubProtocol\\[processed CONNECT\\(0\\)-CONNECTED\\(0\\)-DISCONNECT\\(0\\)\\], " +
|
||||
"stompBrokerRelay\\[null\\], " +
|
||||
"inboundChannel\\[pool size = \\d, active threads = \\d, queued tasks = \\d, completed tasks = \\d\\], " +
|
||||
"outboundChannelpool size = \\d, active threads = \\d, queued tasks = \\d, completed tasks = \\d\\], " +
|
||||
"outboundChannel\\[pool size = \\d, active threads = \\d, queued tasks = \\d, completed tasks = \\d\\], " +
|
||||
"sockJsScheduler\\[pool size = \\d, active threads = \\d, queued tasks = \\d, completed tasks = \\d\\]";
|
||||
|
||||
assertTrue("\nExpected: " + expected.replace("\\", "") + "\n Actual: " + actual, actual.matches(expected));
|
||||
|
||||
Reference in New Issue
Block a user