1 /*
2 * Copyright 2014 The Netty Project
3 *
4 * The Netty Project licenses this file to you under the Apache License,
5 * version 2.0 (the "License"); you may not use this file except in compliance
6 * with the License. You may obtain a copy of the License at:
7 *
8 * https://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 * License for the specific language governing permissions and limitations
14 * under the License.
15 */
16 package io.netty.handler.codec.stomp;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.Unpooled;
20 import io.netty.channel.ChannelHandlerContext;
21 import io.netty.handler.codec.DecoderException;
22 import io.netty.handler.codec.DecoderResult;
23 import io.netty.handler.codec.ReplayingDecoder;
24 import io.netty.handler.codec.TooLongFrameException;
25 import io.netty.handler.codec.stomp.StompSubframeDecoder.State;
26 import io.netty.util.ByteProcessor;
27 import io.netty.util.internal.AppendableCharSequence;
28 import io.netty.util.internal.StringUtil;
29 import io.netty.util.internal.UnstableApi;
30
31 import java.util.List;
32
33 import static io.netty.buffer.ByteBufUtil.*;
34 import static io.netty.util.internal.ObjectUtil.*;
35
36 /**
37 * Decodes {@link ByteBuf}s into {@link StompHeadersSubframe}s and {@link StompContentSubframe}s.
38 *
39 * <h3>Parameters to control memory consumption: </h3>
40 * {@code maxLineLength} the maximum length of line - restricts length of command and header lines If the length of the
41 * initial line exceeds this value, a {@link TooLongFrameException} will be raised.
42 * <br>
43 * {@code maxChunkSize} The maximum length of the content or each chunk. If the content length (or the length of each
44 * chunk) exceeds this value, the content or chunk ill be split into multiple {@link StompContentSubframe}s whose length
45 * is {@code maxChunkSize} at maximum.
46 *
47 * <h3>Chunked Content</h3>
48 * <p>
49 * If the content of a stomp message is greater than {@code maxChunkSize} the transfer encoding of the HTTP message is
50 * 'chunked', this decoder generates multiple {@link StompContentSubframe} instances to avoid excessive memory
51 * consumption. Note, that every message, even with no content decodes with {@link LastStompContentSubframe} at the end
52 * to simplify upstream message parsing.
53 */
54 public class StompSubframeDecoder extends ReplayingDecoder<State> {
55
56 private static final int DEFAULT_CHUNK_SIZE = 8132;
57 private static final int DEFAULT_MAX_LINE_LENGTH = 1024;
58
59 @UnstableApi
60 public enum State {
61 SKIP_CONTROL_CHARACTERS,
62 READ_HEADERS,
63 READ_CONTENT,
64 FINALIZE_FRAME_READ,
65 BAD_FRAME,
66 INVALID_CHUNK
67 }
68
69 private final Utf8LineParser commandParser;
70 private final HeaderParser headerParser;
71 private final int maxChunkSize;
72 private int alreadyReadChunkSize;
73 private LastStompContentSubframe lastContent;
74 private long contentLength = -1;
75
76 public StompSubframeDecoder() {
77 this(DEFAULT_MAX_LINE_LENGTH, DEFAULT_CHUNK_SIZE);
78 }
79
80 public StompSubframeDecoder(boolean validateHeaders) {
81 this(DEFAULT_MAX_LINE_LENGTH, DEFAULT_CHUNK_SIZE, validateHeaders);
82 }
83
84 public StompSubframeDecoder(int maxLineLength, int maxChunkSize) {
85 this(maxLineLength, maxChunkSize, false);
86 }
87
88 public StompSubframeDecoder(int maxLineLength, int maxChunkSize, boolean validateHeaders) {
89 super(State.SKIP_CONTROL_CHARACTERS);
90 checkPositive(maxLineLength, "maxLineLength");
91 checkPositive(maxChunkSize, "maxChunkSize");
92 this.maxChunkSize = maxChunkSize;
93 commandParser = new Utf8LineParser(new AppendableCharSequence(16), maxLineLength);
94 headerParser = new HeaderParser(new AppendableCharSequence(128), maxLineLength, validateHeaders);
95 }
96
97 @Override
98 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
99 switch (state()) {
100 case SKIP_CONTROL_CHARACTERS:
101 skipControlCharacters(in);
102 checkpoint(State.READ_HEADERS);
103 // Fall through.
104 case READ_HEADERS:
105 StompCommand command = StompCommand.UNKNOWN;
106 StompHeadersSubframe frame = null;
107 try {
108 command = readCommand(in);
109 frame = new DefaultStompHeadersSubframe(command);
110 checkpoint(readHeaders(in, frame));
111 out.add(frame);
112 } catch (Exception e) {
113 if (frame == null) {
114 frame = new DefaultStompHeadersSubframe(command);
115 }
116 frame.setDecoderResult(DecoderResult.failure(e));
117 out.add(frame);
118 checkpoint(State.BAD_FRAME);
119 return;
120 }
121 break;
122 case BAD_FRAME:
123 in.skipBytes(actualReadableBytes());
124 return;
125 }
126 try {
127 switch (state()) {
128 case READ_CONTENT:
129 int toRead = in.readableBytes();
130 if (toRead == 0) {
131 return;
132 }
133 if (toRead > maxChunkSize) {
134 toRead = maxChunkSize;
135 }
136 if (contentLength >= 0) {
137 int remainingLength = (int) (contentLength - alreadyReadChunkSize);
138 if (toRead > remainingLength) {
139 toRead = remainingLength;
140 }
141 ByteBuf chunkBuffer = readBytes(ctx.alloc(), in, toRead);
142 if ((alreadyReadChunkSize += toRead) >= contentLength) {
143 lastContent = new DefaultLastStompContentSubframe(chunkBuffer);
144 checkpoint(State.FINALIZE_FRAME_READ);
145 } else {
146 out.add(new DefaultStompContentSubframe(chunkBuffer));
147 return;
148 }
149 } else {
150 int nulIndex = indexOf(in, in.readerIndex(), in.writerIndex(), StompConstants.NUL);
151 if (nulIndex == in.readerIndex()) {
152 checkpoint(State.FINALIZE_FRAME_READ);
153 } else {
154 if (nulIndex > 0) {
155 toRead = nulIndex - in.readerIndex();
156 } else {
157 toRead = in.writerIndex() - in.readerIndex();
158 }
159 ByteBuf chunkBuffer = readBytes(ctx.alloc(), in, toRead);
160 alreadyReadChunkSize += toRead;
161 if (nulIndex > 0) {
162 lastContent = new DefaultLastStompContentSubframe(chunkBuffer);
163 checkpoint(State.FINALIZE_FRAME_READ);
164 } else {
165 out.add(new DefaultStompContentSubframe(chunkBuffer));
166 return;
167 }
168 }
169 }
170 // Fall through.
171 case FINALIZE_FRAME_READ:
172 skipNullCharacter(in);
173 if (lastContent == null) {
174 lastContent = LastStompContentSubframe.EMPTY_LAST_CONTENT;
175 }
176 out.add(lastContent);
177 resetDecoder();
178 }
179 } catch (Exception e) {
180 if (lastContent != null) {
181 lastContent.release();
182 lastContent = null;
183 }
184
185 StompContentSubframe errorContent = new DefaultLastStompContentSubframe(Unpooled.EMPTY_BUFFER);
186 errorContent.setDecoderResult(DecoderResult.failure(e));
187 out.add(errorContent);
188 checkpoint(State.BAD_FRAME);
189 }
190 }
191
192 private StompCommand readCommand(ByteBuf in) {
193 CharSequence commandSequence = commandParser.parse(in);
194 if (commandSequence == null) {
195 throw new DecoderException("Failed to read command from channel");
196 }
197 String commandStr = commandSequence.toString();
198 try {
199 return StompCommand.valueOf(commandStr);
200 } catch (IllegalArgumentException iae) {
201 throw new DecoderException("Cannot to parse command " + commandStr);
202 }
203 }
204
205 private State readHeaders(ByteBuf buffer, StompHeadersSubframe headersSubframe) {
206 StompHeaders headers = headersSubframe.headers();
207 for (;;) {
208 boolean headerRead = headerParser.parseHeader(headersSubframe, buffer);
209 if (!headerRead) {
210 if (headers.contains(StompHeaders.CONTENT_LENGTH)) {
211 contentLength = getContentLength(headers);
212 if (contentLength == 0) {
213 return State.FINALIZE_FRAME_READ;
214 }
215 }
216 return State.READ_CONTENT;
217 }
218 }
219 }
220
221 private static long getContentLength(StompHeaders headers) {
222 long contentLength = headers.getLong(StompHeaders.CONTENT_LENGTH, 0L);
223 if (contentLength < 0) {
224 throw new DecoderException(StompHeaders.CONTENT_LENGTH + " must be non-negative");
225 }
226 return contentLength;
227 }
228
229 private static void skipNullCharacter(ByteBuf buffer) {
230 byte b = buffer.readByte();
231 if (b != StompConstants.NUL) {
232 throw new IllegalStateException("unexpected byte in buffer " + b + " while expecting NULL byte");
233 }
234 }
235
236 private static void skipControlCharacters(ByteBuf buffer) {
237 byte b;
238 for (;;) {
239 b = buffer.readByte();
240 if (b != StompConstants.CR && b != StompConstants.LF) {
241 buffer.readerIndex(buffer.readerIndex() - 1);
242 break;
243 }
244 }
245 }
246
247 private void resetDecoder() {
248 checkpoint(State.SKIP_CONTROL_CHARACTERS);
249 contentLength = -1;
250 alreadyReadChunkSize = 0;
251 lastContent = null;
252 }
253
254 private static class Utf8LineParser implements ByteProcessor {
255
256 private final AppendableCharSequence charSeq;
257 private final int maxLineLength;
258
259 private int lineLength;
260 private char interim;
261 private boolean nextRead;
262
263 Utf8LineParser(AppendableCharSequence charSeq, int maxLineLength) {
264 this.charSeq = checkNotNull(charSeq, "charSeq");
265 this.maxLineLength = maxLineLength;
266 }
267
268 AppendableCharSequence parse(ByteBuf byteBuf) {
269 reset();
270 int offset = byteBuf.forEachByte(this);
271 if (offset == -1) {
272 return null;
273 }
274
275 byteBuf.readerIndex(offset + 1);
276 return charSeq;
277 }
278
279 AppendableCharSequence charSequence() {
280 return charSeq;
281 }
282
283 @Override
284 public boolean process(byte nextByte) throws Exception {
285 if (nextByte == StompConstants.CR) {
286 ++lineLength;
287 return true;
288 }
289
290 if (nextByte == StompConstants.LF) {
291 return false;
292 }
293
294 if (++lineLength > maxLineLength) {
295 throw new TooLongFrameException("An STOMP line is larger than " + maxLineLength + " bytes.");
296 }
297
298 // 1 byte - 0xxxxxxx - 7 bits
299 // 2 byte - 110xxxxx 10xxxxxx - 11 bits
300 // 3 byte - 1110xxxx 10xxxxxx 10xxxxxx - 16 bits
301 if (nextRead) {
302 interim |= (nextByte & 0x3F) << 6;
303 nextRead = false;
304 } else if (interim != 0) { // flush 2 or 3 byte
305 appendTo(charSeq, (char) (interim | (nextByte & 0x3F)));
306 interim = 0;
307 } else if (nextByte >= 0) { // INITIAL BRANCH
308 // The first 128 characters (US-ASCII) need one byte.
309 appendTo(charSeq, (char) nextByte);
310 } else if ((nextByte & 0xE0) == 0xC0) {
311 // The next 1920 characters need two bytes and we can define
312 // a first byte by mask 110xxxxx.
313 interim = (char) ((nextByte & 0x1F) << 6);
314 } else {
315 // The rest of characters need three bytes.
316 interim = (char) ((nextByte & 0x0F) << 12);
317 nextRead = true;
318 }
319
320 return true;
321 }
322
323 protected void appendTo(AppendableCharSequence charSeq, char chr) {
324 charSeq.append(chr);
325 }
326
327 protected void reset() {
328 charSeq.reset();
329 lineLength = 0;
330 interim = 0;
331 nextRead = false;
332 }
333 }
334
335 private static final class HeaderParser extends Utf8LineParser {
336
337 private final boolean validateHeaders;
338
339 private String name;
340 private boolean valid;
341
342 private boolean shouldUnescape;
343 private boolean unescapeInProgress;
344
345 HeaderParser(AppendableCharSequence charSeq, int maxLineLength, boolean validateHeaders) {
346 super(charSeq, maxLineLength);
347 this.validateHeaders = validateHeaders;
348 }
349
350 boolean parseHeader(StompHeadersSubframe headersSubframe, ByteBuf buf) {
351 shouldUnescape = shouldUnescape(headersSubframe.command());
352 AppendableCharSequence value = super.parse(buf);
353 if (value == null || (name == null && value.length() == 0)) {
354 return false;
355 }
356
357 if (valid) {
358 headersSubframe.headers().add(name, value.toString());
359 } else if (validateHeaders) {
360 if (StringUtil.isNullOrEmpty(name)) {
361 throw new IllegalArgumentException("received an invalid header line '" + value + '\'');
362 }
363 String line = name + ':' + value;
364 throw new IllegalArgumentException("a header value or name contains a prohibited character ':'"
365 + ", " + line);
366 }
367 return true;
368 }
369
370 @Override
371 public boolean process(byte nextByte) throws Exception {
372 if (nextByte == StompConstants.COLON) {
373 if (name == null) {
374 AppendableCharSequence charSeq = charSequence();
375 if (charSeq.length() != 0) {
376 name = charSeq.substring(0, charSeq.length());
377 charSeq.reset();
378 valid = true;
379 return true;
380 } else {
381 name = StringUtil.EMPTY_STRING;
382 }
383 } else {
384 valid = false;
385 }
386 }
387
388 return super.process(nextByte);
389 }
390
391 @Override
392 protected void appendTo(AppendableCharSequence charSeq, char chr) {
393 if (!shouldUnescape) {
394 super.appendTo(charSeq, chr);
395 return;
396 }
397
398 if (chr == '\\') {
399 if (unescapeInProgress) {
400 super.appendTo(charSeq, chr);
401 unescapeInProgress = false;
402 } else {
403 unescapeInProgress = true;
404 }
405 return;
406 }
407
408 if (unescapeInProgress) {
409 if (chr == 'c') {
410 charSeq.append(':');
411 } else if (chr == 'r') {
412 charSeq.append('\r');
413 } else if (chr == 'n') {
414 charSeq.append('\n');
415 } else {
416 charSeq.append('\\').append(chr);
417 throw new IllegalArgumentException("received an invalid escape header sequence '" + charSeq + '\'');
418 }
419
420 unescapeInProgress = false;
421 return;
422 }
423
424 super.appendTo(charSeq, chr);
425 }
426
427 @Override
428 protected void reset() {
429 name = null;
430 valid = false;
431 unescapeInProgress = false;
432 super.reset();
433 }
434
435 private static boolean shouldUnescape(StompCommand command) {
436 return command != StompCommand.CONNECT && command != StompCommand.CONNECTED;
437 }
438 }
439 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！