Error handling improvement in AbstractSockJsSession

Backport of e17736d.

Closes gh-24986

spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractSockJsSession.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,15 +17,16 @@
 package org.springframework.web.socket.sockjs.transport.session;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledFuture;
+import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -58,7 +59,7 @@ private enum State {NEW, OPEN, CLOSED}
 
 	/**
 	 * Log category to use on network IO exceptions after a client has gone away.
-	 * <p>Servlet containers dn't expose a a client disconnected callback, see
+	 * <p>Servlet containers don't expose a client disconnected callback; see
 	 * <a href="https://github.com/eclipse-ee4j/servlet-api/issues/44">eclipse-ee4j/servlet-api#44</a>.
 	 * Therefore network IO failures may occur simply because a client has gone away,
 	 * and that can fill the logs with unnecessary stack traces.
@@ -121,7 +122,7 @@ private enum State {NEW, OPEN, CLOSED}
 	 * @param id the session ID
 	 * @param config the SockJS service configuration options
 	 * @param handler the recipient of SockJS messages
-	 * @param attributes attributes from the HTTP handshake to associate with the WebSocket
+	 * @param attributes the attributes from the HTTP handshake to associate with the WebSocket
 	 * session; the provided attributes are copied, the original map is not used.
 	 */
 	public AbstractSockJsSession(String id, SockJsServiceConfig config, WebSocketHandler handler,
@@ -162,6 +163,7 @@ public Map<String, Object> getAttributes() {
 
 	// Message sending
 
+	@Override
 	public final void sendMessage(WebSocketMessage<?> message) throws IOException {
 		Assert.state(!isClosed(), "Cannot send a message when session is closed");
 		Assert.isInstanceOf(TextMessage.class, message, "SockJS supports text messages only");
@@ -376,23 +378,48 @@ public void delegateConnectionEstablished() throws Exception {
 	}
 
 	public void delegateMessages(String... messages) throws SockJsMessageDeliveryException {
-		List<String> undelivered = new ArrayList<>(Arrays.asList(messages));
-		for (String message : messages) {
+		for (int i = 0; i < messages.length; i++) {
 			try {
 				if (isClosed()) {
-					throw new SockJsMessageDeliveryException(this.id, undelivered, "Session closed");
-				}
-				else {
-					this.handler.handleMessage(this, new TextMessage(message));
-					undelivered.remove(0);
+					logUndeliveredMessages(i, messages);
+					return;
 				}
+				this.handler.handleMessage(this, new TextMessage(messages[i]));
 			}
-			catch (Throwable ex) {
-				throw new SockJsMessageDeliveryException(this.id, undelivered, ex);
+			catch (Exception ex) {
+				if (isClosed()) {
+					if (logger.isTraceEnabled()) {
+						logger.trace("Failed to handle message '" + messages[i] + "'", ex);
+					}
+					logUndeliveredMessages(i, messages);
+					return;
+				}
+				throw new SockJsMessageDeliveryException(this.id, getUndelivered(messages, i), ex);
 			}
 		}
 	}
 
+	private void logUndeliveredMessages(int index, String[] messages) {
+		List<String> undelivered = getUndelivered(messages, index);
+		if (logger.isTraceEnabled() && !undelivered.isEmpty()) {
+			logger.trace("Dropped inbound message(s) due to closed session: " + undelivered);
+		}
+	}
+
+	private static List<String> getUndelivered(String[] messages, int i) {
+		switch (messages.length - i) {
+			case 0:
+				return Collections.emptyList();
+			case 1:
+				return (messages[i].trim().isEmpty() ?
+						Collections.emptyList() : Collections.singletonList(messages[i]));
+			default:
+				return Arrays.stream(Arrays.copyOfRange(messages, i, messages.length))
+						.filter(message -> !message.trim().isEmpty())
+						.collect(Collectors.toList());
+		}
+	}
+
 	/**
 	 * Invoked when the underlying connection is closed.
 	 */

spring-websocket/src/test/java/org/springframework/web/socket/sockjs/transport/session/SockJsSessionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,7 +27,6 @@
 import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketHandler;
 import org.springframework.web.socket.handler.ExceptionWebSocketHandlerDecorator;
-import org.springframework.web.socket.sockjs.SockJsMessageDeliveryException;
 import org.springframework.web.socket.sockjs.SockJsTransportFailureException;
 import org.springframework.web.socket.sockjs.frame.SockJsFrame;
 
@@ -118,18 +117,13 @@ public void delegateMessagesWithErrorAndConnectionClosing() throws Exception {
 		willThrow(new IOException()).given(this.webSocketHandler).handleMessage(sockJsSession, new TextMessage(msg2));
 
 		sockJsSession.delegateConnectionEstablished();
-		try {
-			sockJsSession.delegateMessages(msg1, msg2, msg3);
-			fail("expected exception");
-		}
-		catch (SockJsMessageDeliveryException ex) {
-			assertEquals(Collections.singletonList(msg3), ex.getUndeliveredMessages());
-			verify(this.webSocketHandler).afterConnectionEstablished(sockJsSession);
-			verify(this.webSocketHandler).handleMessage(sockJsSession, new TextMessage(msg1));
-			verify(this.webSocketHandler).handleMessage(sockJsSession, new TextMessage(msg2));
-			verify(this.webSocketHandler).afterConnectionClosed(sockJsSession, CloseStatus.SERVER_ERROR);
-			verifyNoMoreInteractions(this.webSocketHandler);
-		}
+		sockJsSession.delegateMessages(msg1, msg2, msg3);
+
+		verify(this.webSocketHandler).afterConnectionEstablished(sockJsSession);
+		verify(this.webSocketHandler).handleMessage(sockJsSession, new TextMessage(msg1));
+		verify(this.webSocketHandler).handleMessage(sockJsSession, new TextMessage(msg2));
+		verify(this.webSocketHandler).afterConnectionClosed(sockJsSession, CloseStatus.SERVER_ERROR);
+		verifyNoMoreInteractions(this.webSocketHandler);
 	}
 
 	@Test