diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java b/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java index b0efe1753d..f52bae7a5f 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecorator.java @@ -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. @@ -31,13 +31,14 @@ import org.springframework.web.socket.WebSocketMessage; import org.springframework.web.socket.WebSocketSession; /** - * Wraps a {@link org.springframework.web.socket.WebSocketSession} and guarantees - * only one thread can send messages at a time. + * Wrap a {@link org.springframework.web.socket.WebSocketSession WebSocketSession} + * to guarantee only one thread can send messages at a time. * - *

If a send is slow, subsequent attempts to send more messages from a different - * thread will fail to acquire the flush lock and the messages will be buffered - * instead: At that time, the specified buffer-size limit and send-time limit will - * be checked and the session closed if the limits are exceeded. + *

If a send is slow, subsequent attempts to send more messages from other + * threads will not be able to acquire the flush lock and messages will be + * buffered instead -- at that time, the specified buffer-size limit and + * send-time limit will be checked and the session closed if the limits are + * exceeded. * * @author Rossen Stoyanchev * @since 4.0.3 @@ -47,21 +48,20 @@ public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorat private static final Log logger = LogFactory.getLog(ConcurrentWebSocketSessionDecorator.class); - private final Queue> buffer = new LinkedBlockingQueue>(); - - private final AtomicInteger bufferSize = new AtomicInteger(); + private final int sendTimeLimit; private final int bufferSizeLimit; + private final Queue> buffer = new LinkedBlockingQueue>(); + + private final AtomicInteger bufferSize = new AtomicInteger(); + private volatile long sendStartTime; - private final int sendTimeLimit; - - private volatile boolean limitExceeded; - private volatile boolean shutdownInProgress; + private volatile boolean closeInProgress; private final Lock flushLock = new ReentrantLock(); @@ -93,7 +93,7 @@ public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorat public void sendMessage(WebSocketMessage message) throws IOException { - if (isDisabled()) { + if (shouldNotSend()) { return; } @@ -103,32 +103,33 @@ public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorat do { if (!tryFlushMessageBuffer()) { if (logger.isTraceEnabled()) { - logger.trace("Another send already in progress, session id '" + - getId() + "'" + ", in-progress send time " + getTimeSinceSendStarted() + - " (ms)" + ", buffer size " + this.bufferSize + " bytes"); + String text = String.format("Another send already in progress: " + + "session id '%s':, \"in-progress\" send time %d (ms), buffer size %d bytes", + getId(), getTimeSinceSendStarted(), this.bufferSize.get()); + logger.trace(text); } checkSessionLimits(); break; } } - while (!this.buffer.isEmpty() && !isDisabled()); + while (!this.buffer.isEmpty() && !shouldNotSend()); } - private boolean isDisabled() { - return (this.limitExceeded || this.shutdownInProgress); + private boolean shouldNotSend() { + return (this.limitExceeded || this.closeInProgress); } private boolean tryFlushMessageBuffer() throws IOException { if (this.flushLock.tryLock()) { try { while (true) { - WebSocketMessage messageToSend = this.buffer.poll(); - if (messageToSend == null || isDisabled()) { + WebSocketMessage message = this.buffer.poll(); + if (message == null || shouldNotSend()) { break; } - this.bufferSize.addAndGet(messageToSend.getPayloadLength() * -1); + this.bufferSize.addAndGet(message.getPayloadLength() * -1); this.sendStartTime = System.currentTimeMillis(); - getDelegate().sendMessage(messageToSend); + getDelegate().sendMessage(message); this.sendStartTime = 0; } } @@ -142,18 +143,17 @@ public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorat } private void checkSessionLimits() throws IOException { - if (!isDisabled() && this.closeLock.tryLock()) { + if (!shouldNotSend() && this.closeLock.tryLock()) { try { if (getTimeSinceSendStarted() > this.sendTimeLimit) { - String errorMessage = "Message send time " + getTimeSinceSendStarted() + - " (ms) exceeded the allowed limit " + this.sendTimeLimit; - sessionLimitReached(errorMessage, CloseStatus.SESSION_NOT_RELIABLE); + String format = "Message send time %d (ms) for session '%s' exceeded the allowed limit %d"; + String reason = String.format(format, getTimeSinceSendStarted(), getId(), this.sendTimeLimit); + setLimitExceeded(reason); } else if (this.bufferSize.get() > this.bufferSizeLimit) { - String errorMessage = "The send buffer size " + this.bufferSize.get() + " bytes for " + - "session '" + getId() + " exceeded the allowed limit " + this.bufferSizeLimit; - sessionLimitReached(errorMessage, - (getTimeSinceSendStarted() >= 10000 ? CloseStatus.SESSION_NOT_RELIABLE : null)); + String format = "The send buffer size %d bytes for session '%s' exceeded the allowed limit %d"; + String reason = String.format(format, this.bufferSize.get(), getId(), this.bufferSizeLimit); + setLimitExceeded(reason); } } finally { @@ -162,14 +162,14 @@ public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorat } } - private void sessionLimitReached(String reason, CloseStatus status) { + private void setLimitExceeded(String reason) { this.limitExceeded = true; - throw new SessionLimitExceededException(reason, status); + throw new SessionLimitExceededException(reason, CloseStatus.SESSION_NOT_RELIABLE); } @Override public void close(CloseStatus status) throws IOException { - this.shutdownInProgress = true; + this.closeInProgress = true; super.close(status); } diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java b/spring-websocket/src/test/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java index 7e73979ab3..6523624e1f 100644 --- a/spring-websocket/src/test/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java +++ b/spring-websocket/src/test/java/org/springframework/web/socket/handler/ConcurrentWebSocketSessionDecoratorTests.java @@ -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. @@ -39,6 +39,7 @@ import static org.junit.Assert.*; @SuppressWarnings("resource") public class ConcurrentWebSocketSessionDecoratorTests { + @Test public void send() throws IOException { @@ -69,16 +70,13 @@ public class ConcurrentWebSocketSessionDecoratorTests { final ConcurrentWebSocketSessionDecorator concurrentSession = new ConcurrentWebSocketSessionDecorator(blockingSession, 10 * 1000, 1024); - Executors.newSingleThreadExecutor().submit(new Runnable() { - @Override - public void run() { - TextMessage textMessage = new TextMessage("slow message"); - try { - concurrentSession.sendMessage(textMessage); - } - catch (IOException e) { - e.printStackTrace(); - } + Executors.newSingleThreadExecutor().submit((Runnable) () -> { + TextMessage message = new TextMessage("slow message"); + try { + concurrentSession.sendMessage(message); + } + catch (IOException e) { + e.printStackTrace(); } }); @@ -102,6 +100,7 @@ public class ConcurrentWebSocketSessionDecoratorTests { public void sendTimeLimitExceeded() throws IOException, InterruptedException { BlockingSession blockingSession = new BlockingSession(); + blockingSession.setId("123"); blockingSession.setOpen(true); CountDownLatch sentMessageLatch = blockingSession.getSentMessageLatch(); @@ -111,16 +110,13 @@ public class ConcurrentWebSocketSessionDecoratorTests { final ConcurrentWebSocketSessionDecorator concurrentSession = new ConcurrentWebSocketSessionDecorator(blockingSession, sendTimeLimit, bufferSizeLimit); - Executors.newSingleThreadExecutor().submit(new Runnable() { - @Override - public void run() { - TextMessage textMessage = new TextMessage("slow message"); - try { - concurrentSession.sendMessage(textMessage); - } - catch (IOException e) { - e.printStackTrace(); - } + Executors.newSingleThreadExecutor().submit((Runnable) () -> { + TextMessage message = new TextMessage("slow message"); + try { + concurrentSession.sendMessage(message); + } + catch (IOException e) { + e.printStackTrace(); } }); @@ -135,6 +131,9 @@ public class ConcurrentWebSocketSessionDecoratorTests { fail("Expected exception"); } catch (SessionLimitExceededException ex) { + String actual = ex.getMessage(); + String regex = "Message send time [\\d]+ \\(ms\\) for session '123' exceeded the allowed limit 100"; + assertTrue("Unexpected message: " + actual, actual.matches(regex)); assertEquals(CloseStatus.SESSION_NOT_RELIABLE, ex.getStatus()); } } @@ -143,6 +142,7 @@ public class ConcurrentWebSocketSessionDecoratorTests { public void sendBufferSizeExceeded() throws IOException, InterruptedException { BlockingSession blockingSession = new BlockingSession(); + blockingSession.setId("123"); blockingSession.setOpen(true); CountDownLatch sentMessageLatch = blockingSession.getSentMessageLatch(); @@ -152,16 +152,13 @@ public class ConcurrentWebSocketSessionDecoratorTests { final ConcurrentWebSocketSessionDecorator concurrentSession = new ConcurrentWebSocketSessionDecorator(blockingSession, sendTimeLimit, bufferSizeLimit); - Executors.newSingleThreadExecutor().submit(new Runnable() { - @Override - public void run() { - TextMessage textMessage = new TextMessage("slow message"); - try { - concurrentSession.sendMessage(textMessage); - } - catch (IOException e) { - e.printStackTrace(); - } + Executors.newSingleThreadExecutor().submit((Runnable) () -> { + TextMessage message = new TextMessage("slow message"); + try { + concurrentSession.sendMessage(message); + } + catch (IOException e) { + e.printStackTrace(); } }); @@ -183,6 +180,10 @@ public class ConcurrentWebSocketSessionDecoratorTests { fail("Expected exception"); } catch (SessionLimitExceededException ex) { + String actual = ex.getMessage(); + String regex = "The send buffer size [\\d]+ bytes for session '123' exceeded the allowed limit 1024"; + assertTrue("Unexpected message: " + actual, actual.matches(regex)); + assertEquals(CloseStatus.SESSION_NOT_RELIABLE, ex.getStatus()); } } @@ -220,47 +221,4 @@ public class ConcurrentWebSocketSessionDecoratorTests { } -// @Test -// public void sendSessionLimitException() throws IOException, InterruptedException { -// -// BlockingSession blockingSession = new BlockingSession(); -// blockingSession.setOpen(true); -// CountDownLatch sentMessageLatch = blockingSession.getSentMessageLatch(); -// -// int sendTimeLimit = 10 * 1000; -// int bufferSizeLimit = 1024; -// -// final ConcurrentWebSocketSessionDecorator concurrentSession = -// new ConcurrentWebSocketSessionDecorator(blockingSession, sendTimeLimit, bufferSizeLimit); -// -// Executors.newSingleThreadExecutor().submit(new Runnable() { -// @Override -// public void run() { -// TextMessage textMessage = new TextMessage("slow message"); -// try { -// concurrentSession.sendMessage(textMessage); -// } -// catch (IOException e) { -// e.printStackTrace(); -// } -// } -// }); -// -// assertTrue(sentMessageLatch.await(5, TimeUnit.SECONDS)); -// -// StringBuilder sb = new StringBuilder(); -// for (int i=0 ; i < 1023; i++) { -// sb.append("a"); -// } -// -// TextMessage message = new TextMessage(sb.toString()); -// concurrentSession.sendMessage(message); -// -// assertEquals(1023, concurrentSession.getBufferSize()); -// assertTrue(blockingSession.isOpen()); -// -// concurrentSession.sendMessage(message); -// assertFalse(blockingSession.isOpen()); -// } - }