1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.handler.codec.stomp;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.ByteBufUtil;
20 import io.netty.channel.ChannelHandlerContext;
21 import io.netty.handler.codec.MessageToMessageEncoder;
22 import io.netty.util.concurrent.FastThreadLocal;
23 import io.netty.util.internal.AppendableCharSequence;
24 import io.netty.util.internal.PlatformDependent;
25
26 import java.util.LinkedHashMap;
27 import java.util.List;
28 import java.util.Map.Entry;
29
30 import static io.netty.handler.codec.stomp.StompConstants.NUL;
31 import static io.netty.handler.codec.stomp.StompHeaders.ACCEPT_VERSION;
32 import static io.netty.handler.codec.stomp.StompHeaders.ACK;
33 import static io.netty.handler.codec.stomp.StompHeaders.CONTENT_LENGTH;
34 import static io.netty.handler.codec.stomp.StompHeaders.CONTENT_TYPE;
35 import static io.netty.handler.codec.stomp.StompHeaders.DESTINATION;
36 import static io.netty.handler.codec.stomp.StompHeaders.HEART_BEAT;
37 import static io.netty.handler.codec.stomp.StompHeaders.HOST;
38 import static io.netty.handler.codec.stomp.StompHeaders.ID;
39 import static io.netty.handler.codec.stomp.StompHeaders.LOGIN;
40 import static io.netty.handler.codec.stomp.StompHeaders.MESSAGE;
41 import static io.netty.handler.codec.stomp.StompHeaders.MESSAGE_ID;
42 import static io.netty.handler.codec.stomp.StompHeaders.PASSCODE;
43 import static io.netty.handler.codec.stomp.StompHeaders.RECEIPT;
44 import static io.netty.handler.codec.stomp.StompHeaders.RECEIPT_ID;
45 import static io.netty.handler.codec.stomp.StompHeaders.SERVER;
46 import static io.netty.handler.codec.stomp.StompHeaders.SESSION;
47 import static io.netty.handler.codec.stomp.StompHeaders.SUBSCRIPTION;
48 import static io.netty.handler.codec.stomp.StompHeaders.TRANSACTION;
49 import static io.netty.handler.codec.stomp.StompHeaders.VERSION;
50
51
52
53
54 public class StompSubframeEncoder extends MessageToMessageEncoder<StompSubframe> {
55
56 private static final int ESCAPE_HEADER_KEY_CACHE_LIMIT = 32;
57 private static final float DEFAULT_LOAD_FACTOR = 0.75f;
58 private static final FastThreadLocal<LinkedHashMap<CharSequence, CharSequence>> ESCAPE_HEADER_KEY_CACHE =
59 new FastThreadLocal<LinkedHashMap<CharSequence, CharSequence>>() {
60 @Override
61 protected LinkedHashMap<CharSequence, CharSequence> initialValue() throws Exception {
62 LinkedHashMap<CharSequence, CharSequence> cache = new LinkedHashMap<CharSequence, CharSequence>(
63 ESCAPE_HEADER_KEY_CACHE_LIMIT, DEFAULT_LOAD_FACTOR, true) {
64
65 @Override
66 protected boolean removeEldestEntry(Entry eldest) {
67 return size() > ESCAPE_HEADER_KEY_CACHE_LIMIT;
68 }
69 };
70
71 cache.put(ACCEPT_VERSION, ACCEPT_VERSION);
72 cache.put(HOST, HOST);
73 cache.put(LOGIN, LOGIN);
74 cache.put(PASSCODE, PASSCODE);
75 cache.put(HEART_BEAT, HEART_BEAT);
76 cache.put(VERSION, VERSION);
77 cache.put(SESSION, SESSION);
78 cache.put(SERVER, SERVER);
79 cache.put(DESTINATION, DESTINATION);
80 cache.put(ID, ID);
81 cache.put(ACK, ACK);
82 cache.put(TRANSACTION, TRANSACTION);
83 cache.put(RECEIPT, RECEIPT);
84 cache.put(MESSAGE_ID, MESSAGE_ID);
85 cache.put(SUBSCRIPTION, SUBSCRIPTION);
86 cache.put(RECEIPT_ID, RECEIPT_ID);
87 cache.put(MESSAGE, MESSAGE);
88 cache.put(CONTENT_LENGTH, CONTENT_LENGTH);
89 cache.put(CONTENT_TYPE, CONTENT_TYPE);
90
91 return cache;
92 }
93 };
94
95 @Override
96 protected void encode(ChannelHandlerContext ctx, StompSubframe msg, List<Object> out) throws Exception {
97 if (msg instanceof StompFrame) {
98 StompFrame stompFrame = (StompFrame) msg;
99 ByteBuf buf = encodeFullFrame(stompFrame, ctx);
100
101 out.add(convertFullFrame(stompFrame, buf));
102 } else if (msg instanceof StompHeadersSubframe) {
103 StompHeadersSubframe stompHeadersSubframe = (StompHeadersSubframe) msg;
104 ByteBuf buf = ctx.alloc().buffer(headersSubFrameSize(stompHeadersSubframe));
105 try {
106 encodeHeaders(stompHeadersSubframe, buf);
107 } catch (Exception e) {
108 buf.release();
109 PlatformDependent.throwException(e);
110 }
111
112 out.add(convertHeadersSubFrame(stompHeadersSubframe, buf));
113 } else if (msg instanceof StompContentSubframe) {
114 StompContentSubframe stompContentSubframe = (StompContentSubframe) msg;
115 ByteBuf buf = encodeContent(stompContentSubframe, ctx);
116
117 out.add(convertContentSubFrame(stompContentSubframe, buf));
118 }
119 }
120
121
122
123
124
125
126
127 protected Object convertFullFrame(StompFrame original, ByteBuf encoded) {
128 return encoded;
129 }
130
131
132
133
134
135
136
137 protected Object convertHeadersSubFrame(StompHeadersSubframe original, ByteBuf encoded) {
138 return encoded;
139 }
140
141
142
143
144
145
146
147 protected Object convertContentSubFrame(StompContentSubframe original, ByteBuf encoded) {
148 return encoded;
149 }
150
151
152
153
154
155 protected int headersSubFrameSize(StompHeadersSubframe headersSubframe) {
156 int estimatedSize = headersSubframe.headers().size() * 34 + 48;
157 if (estimatedSize < 128) {
158 return 128;
159 }
160
161 return Math.max(estimatedSize, 256);
162 }
163
164 private ByteBuf encodeFullFrame(StompFrame frame, ChannelHandlerContext ctx) {
165 int contentReadableBytes = frame.content().readableBytes();
166 ByteBuf buf = ctx.alloc().buffer(headersSubFrameSize(frame) + contentReadableBytes);
167 try {
168 encodeHeaders(frame, buf);
169 } catch (Exception e) {
170 buf.release();
171 PlatformDependent.throwException(e);
172 }
173
174 if (contentReadableBytes > 0) {
175 buf.writeBytes(frame.content());
176 }
177
178 return buf.writeByte(NUL);
179 }
180
181 private static void encodeHeaders(StompHeadersSubframe frame, ByteBuf buf) {
182 StompCommand command = frame.command();
183 ByteBufUtil.writeUtf8(buf, command.toString());
184 buf.writeByte(StompConstants.LF);
185
186 boolean shouldEscape = shouldEscape(command);
187 LinkedHashMap<CharSequence, CharSequence> cache = ESCAPE_HEADER_KEY_CACHE.get();
188 for (Entry<CharSequence, CharSequence> entry : frame.headers()) {
189 CharSequence headerKey = entry.getKey();
190 CharSequence headerValue = entry.getValue();
191 if (headerKey.length() == 0) {
192 throw new IllegalArgumentException("STOMP " + command + " contains empty header name");
193 }
194 if (shouldEscape) {
195 CharSequence cachedHeaderKey = cache.get(headerKey);
196 if (cachedHeaderKey == null) {
197 cachedHeaderKey = escape(command, "header name", headerKey);
198 cache.put(headerKey, cachedHeaderKey);
199 }
200 headerKey = cachedHeaderKey;
201 headerValue = escape(command, "header value", headerValue);
202 } else {
203
204 validateNoIllegalCharacters(command, headerKey, "header name");
205 validateNoIllegalCharacters(command, headerValue, "header value");
206 }
207
208 ByteBufUtil.writeUtf8(buf, headerKey);
209 buf.writeByte(StompConstants.COLON);
210
211 ByteBufUtil.writeUtf8(buf, headerValue);
212 buf.writeByte(StompConstants.LF);
213 }
214
215 buf.writeByte(StompConstants.LF);
216 }
217
218 private static ByteBuf encodeContent(StompContentSubframe content, ChannelHandlerContext ctx) {
219 if (content instanceof LastStompContentSubframe) {
220 ByteBuf buf = ctx.alloc().buffer(content.content().readableBytes() + 1);
221 buf.writeBytes(content.content());
222 buf.writeByte(StompConstants.NUL);
223 return buf;
224 }
225
226 return content.content().retain();
227 }
228
229 private static boolean shouldEscape(StompCommand command) {
230 return command != StompCommand.CONNECT && command != StompCommand.CONNECTED;
231 }
232
233 private static void validateNoIllegalCharacters(StompCommand command, CharSequence value, String type) {
234 for (int i = 0; i < value.length(); i++) {
235 char c = value.charAt(i);
236 if (c == '\n' || c == '\r' || c == ':' || c == '\0') {
237 throw newIllegalCharacterException(command, type, i);
238 }
239 }
240 }
241
242 private static IllegalArgumentException newIllegalCharacterException(StompCommand command, String type, int index) {
243 return new IllegalArgumentException(
244 "STOMP " + command + " " + type + " contains illegal character at index " + index);
245 }
246
247 private static CharSequence escape(StompCommand command, String type, CharSequence input) {
248 AppendableCharSequence builder = null;
249 for (int i = 0; i < input.length(); i++) {
250 char chr = input.charAt(i);
251 if (chr == '\\') {
252 builder = escapeBuilder(builder, input, i);
253 builder.append("\\\\");
254 } else if (chr == ':') {
255 builder = escapeBuilder(builder, input, i);
256 builder.append("\\c");
257 } else if (chr == '\n') {
258 builder = escapeBuilder(builder, input, i);
259 builder.append("\\n");
260 } else if (chr == '\r') {
261 builder = escapeBuilder(builder, input, i);
262 builder.append("\\r");
263 } else if (chr == '\0') {
264
265 throw newIllegalCharacterException(command, type, i);
266 } else if (builder != null) {
267 builder.append(chr);
268 }
269 }
270
271 return builder != null? builder : input;
272 }
273
274 private static AppendableCharSequence escapeBuilder(AppendableCharSequence builder, CharSequence input,
275 int offset) {
276 if (builder != null) {
277 return builder;
278 }
279
280
281 return new AppendableCharSequence(input.length() + 8).append(input, 0, offset);
282 }
283 }