From 9b76dc2ab4364e9082b010ef71b6e08ebe495e2e Mon Sep 17 00:00:00 2001 From: Christoph Dreis Date: Fri, 11 Nov 2016 21:41:25 +0100 Subject: [PATCH 1/2] SPR-14901 Allow customization of STOMP message header encoding Fixes SPR-14901 --- .../messaging/simp/stomp/StompEncoder.java | 53 ++++++++++++++++--- .../WebMvcStompEndpointRegistry.java | 1 - .../messaging/StompSubProtocolHandler.java | 27 ++++++++-- 3 files changed, 69 insertions(+), 12 deletions(-) diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java index 67e2088636..1654a9f3e1 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java @@ -21,6 +21,7 @@ import java.io.DataOutputStream; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -50,6 +51,16 @@ public class StompEncoder { private static final Log logger = LogFactory.getLog(StompEncoder.class); + private static final int HEADER_KEY_CACHE_LIMIT = 32; + + @SuppressWarnings("serial") + private final Map headerKeyCache = + new LinkedHashMap(HEADER_KEY_CACHE_LIMIT, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry eldest) { + return size() > HEADER_KEY_CACHE_LIMIT; + } + }; /** * Encodes the given STOMP {@code message} into a {@code byte[]} @@ -130,11 +141,11 @@ public class StompEncoder { values = Collections.singletonList(StompHeaderAccessor.getPasscode(headers)); } - byte[] encodedKey = encodeHeaderString(entry.getKey(), shouldEscape); + byte[] encodedKey = encodeHeaderKey(entry.getKey(), shouldEscape); for (String value : values) { output.write(encodedKey); output.write(COLON); - output.write(encodeHeaderString(value, shouldEscape)); + output.write(encodeHeaderValue(value, shouldEscape)); output.write(LF); } } @@ -147,9 +158,23 @@ public class StompEncoder { } } - private byte[] encodeHeaderString(String input, boolean escape) { + private byte[] encodeHeaderKey(String input, boolean escape) { String inputToUse = (escape ? escape(input) : input); - return inputToUse.getBytes(StandardCharsets.UTF_8); + if (headerKeyCache.containsKey(inputToUse)) { + return headerKeyCache.get(inputToUse); + } + byte[] bytes = encodeHeaderString(inputToUse); + headerKeyCache.put(inputToUse, bytes); + return bytes; + } + + private byte[] encodeHeaderValue(String input, boolean escape) { + String inputToUse = (escape ? escape(input) : input); + return encodeHeaderString(inputToUse); + } + + private byte[] encodeHeaderString(String input) { + return input.getBytes(StandardCharsets.UTF_8); } /** @@ -157,26 +182,38 @@ public class StompEncoder { * "Value Encoding". */ private String escape(String inString) { - StringBuilder sb = new StringBuilder(inString.length()); + StringBuilder sb = null; for (int i = 0; i < inString.length(); i++) { char c = inString.charAt(i); if (c == '\\') { + sb = getStringBuilder(sb, inString, i); sb.append("\\\\"); } else if (c == ':') { + sb = getStringBuilder(sb, inString, i); sb.append("\\c"); } else if (c == '\n') { - sb.append("\\n"); + sb = getStringBuilder(sb, inString, i); + sb.append("\\n"); } else if (c == '\r') { + sb = getStringBuilder(sb, inString, i); sb.append("\\r"); } - else { + else if (sb != null){ sb.append(c); } } - return sb.toString(); + return (sb != null ? sb.toString() : inString); + } + + private StringBuilder getStringBuilder(StringBuilder sb, String inString, int i) { + if (sb == null) { + sb = new StringBuilder(inString.length()); + sb.append(inString.substring(0, i)); + } + return sb; } private void writeBody(byte[] payload, DataOutputStream output) throws IOException { diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebMvcStompEndpointRegistry.java b/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebMvcStompEndpointRegistry.java index e30ddfb9e0..d3cb5b1a8e 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebMvcStompEndpointRegistry.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebMvcStompEndpointRegistry.java @@ -142,7 +142,6 @@ public class WebMvcStompEndpointRegistry implements StompEndpointRegistry { this.stompHandler.setApplicationEventPublisher(applicationContext); } - /** * Return a handler mapping with the mapped ViewControllers; or {@code null} * in case of no registrations. diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java index 3c0408bb96..06cc1c4849 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java @@ -88,14 +88,13 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE private static final byte[] EMPTY_PAYLOAD = new byte[0]; - private StompSubProtocolErrorHandler errorHandler; private int messageSizeLimit = 64 * 1024; - private final StompEncoder stompEncoder = new StompEncoder(); + private StompEncoder stompEncoder; - private final StompDecoder stompDecoder = new StompDecoder(); + private StompDecoder stompDecoder; private final Map decoders = new ConcurrentHashMap<>(); @@ -107,6 +106,10 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE private final Stats stats = new Stats(); + public StompSubProtocolHandler() { + setEncoder(new StompEncoder()); + setDecoder(new StompDecoder()); + } /** * Configure a handler for error messages sent to clients which allows @@ -126,6 +129,24 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE return this.errorHandler; } + /** + * Configure a {@link StompEncoder} for encoding STOMP frames + * @param encoder the encoder + * @since 4.3.5 + */ + public void setEncoder(StompEncoder encoder) { + this.stompEncoder = encoder; + } + + /** + * Configure a {@link StompDecoder} for decoding STOMP frames + * @param decoder the decoder + * @since 4.3.5 + */ + public void setDecoder(StompDecoder decoder) { + this.stompDecoder = decoder; + } + /** * Configure the maximum size allowed for an incoming STOMP message. * Since a STOMP message can be received in multiple WebSocket messages, From b3fa1b40a0bb1644aa26c0778070026702574a5e Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Thu, 8 Dec 2016 11:13:08 -0500 Subject: [PATCH 2/2] Synchronized updates of STOMP header key cache Issue: SPR-14901 --- .../messaging/simp/stomp/StompEncoder.java | 38 +++++++++++----- .../messaging/StompSubProtocolHandler.java | 45 +++++++++---------- 2 files changed, 47 insertions(+), 36 deletions(-) diff --git a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java index 1654a9f3e1..53e747c378 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompEncoder.java @@ -25,6 +25,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.concurrent.ConcurrentHashMap; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -53,15 +54,26 @@ public class StompEncoder { private static final int HEADER_KEY_CACHE_LIMIT = 32; + + private final Map headerKeyAccessCache = + new ConcurrentHashMap<>(HEADER_KEY_CACHE_LIMIT); + @SuppressWarnings("serial") - private final Map headerKeyCache = + private final Map headerKeyUpdateCache = new LinkedHashMap(HEADER_KEY_CACHE_LIMIT, 0.75f, true) { @Override protected boolean removeEldestEntry(Map.Entry eldest) { - return size() > HEADER_KEY_CACHE_LIMIT; + if (size() > HEADER_KEY_CACHE_LIMIT) { + headerKeyAccessCache.remove(eldest.getKey()); + return true; + } + else { + return false; + } } }; + /** * Encodes the given STOMP {@code message} into a {@code byte[]} * @param message the message to encode @@ -160,21 +172,23 @@ public class StompEncoder { private byte[] encodeHeaderKey(String input, boolean escape) { String inputToUse = (escape ? escape(input) : input); - if (headerKeyCache.containsKey(inputToUse)) { - return headerKeyCache.get(inputToUse); + if (this.headerKeyAccessCache.containsKey(inputToUse)) { + return this.headerKeyAccessCache.get(inputToUse); + } + synchronized (this.headerKeyUpdateCache) { + byte[] bytes = this.headerKeyUpdateCache.get(inputToUse); + if (bytes == null) { + bytes = inputToUse.getBytes(StandardCharsets.UTF_8); + this.headerKeyAccessCache.put(inputToUse, bytes); + this.headerKeyUpdateCache.put(inputToUse, bytes); + } + return bytes; } - byte[] bytes = encodeHeaderString(inputToUse); - headerKeyCache.put(inputToUse, bytes); - return bytes; } private byte[] encodeHeaderValue(String input, boolean escape) { String inputToUse = (escape ? escape(input) : input); - return encodeHeaderString(inputToUse); - } - - private byte[] encodeHeaderString(String input) { - return input.getBytes(StandardCharsets.UTF_8); + return inputToUse.getBytes(StandardCharsets.UTF_8); } /** diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java index 06cc1c4849..19bd622031 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java @@ -88,13 +88,14 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE private static final byte[] EMPTY_PAYLOAD = new byte[0]; + private StompSubProtocolErrorHandler errorHandler; private int messageSizeLimit = 64 * 1024; - private StompEncoder stompEncoder; + private StompEncoder stompEncoder = new StompEncoder(); - private StompDecoder stompDecoder; + private StompDecoder stompDecoder = new StompDecoder(); private final Map decoders = new ConcurrentHashMap<>(); @@ -106,10 +107,6 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE private final Stats stats = new Stats(); - public StompSubProtocolHandler() { - setEncoder(new StompEncoder()); - setDecoder(new StompDecoder()); - } /** * Configure a handler for error messages sent to clients which allows @@ -129,24 +126,6 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE return this.errorHandler; } - /** - * Configure a {@link StompEncoder} for encoding STOMP frames - * @param encoder the encoder - * @since 4.3.5 - */ - public void setEncoder(StompEncoder encoder) { - this.stompEncoder = encoder; - } - - /** - * Configure a {@link StompDecoder} for decoding STOMP frames - * @param decoder the decoder - * @since 4.3.5 - */ - public void setDecoder(StompDecoder decoder) { - this.stompDecoder = decoder; - } - /** * Configure the maximum size allowed for an incoming STOMP message. * Since a STOMP message can be received in multiple WebSocket messages, @@ -167,6 +146,24 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE return this.messageSizeLimit; } + /** + * Configure a {@link StompEncoder} for encoding STOMP frames + * @param encoder the encoder + * @since 4.3.5 + */ + public void setEncoder(StompEncoder encoder) { + this.stompEncoder = encoder; + } + + /** + * Configure a {@link StompDecoder} for decoding STOMP frames + * @param decoder the decoder + * @since 4.3.5 + */ + public void setDecoder(StompDecoder decoder) { + this.stompDecoder = decoder; + } + /** * Configure a {@link MessageHeaderInitializer} to apply to the headers of all * messages created from decoded STOMP frames and other messages sent to the