hunt.http.codec.http.decode.GoAwayBodyParser source code

1 module hunt.http.codec.http.decode.GoAwayBodyParser;
2 
3 import hunt.io.ByteBuffer;
4 
5 import hunt.http.codec.http.decode.BodyParser;
6 import hunt.http.codec.http.decode.HeaderParser;
7 import hunt.http.codec.http.decode.Parser;
8 
9 import hunt.http.codec.http.frame.ErrorCode;
10 import hunt.http.codec.http.frame.GoAwayFrame;
11 
12 import hunt.Exceptions;
13 
14 /**
15 */
16 class GoAwayBodyParser :BodyParser {
17 	private State state = State.PREPARE;
18 	private int cursor;
19 	private int length;
20 	private int lastStreamId;
21 	private int error;
22 	private byte[] payload;
23 
24 	this(HeaderParser headerParser, Parser.Listener listener) {
25 		super(headerParser, listener);
26 	}
27 
28 	private void reset() {
29 		state = State.PREPARE;
30 		cursor = 0;
31 		length = 0;
32 		lastStreamId = 0;
33 		error = 0;
34 		payload = null;
35 	}
36 
37 	override
38 	bool parse(ByteBuffer buffer) {
39 		while (buffer.hasRemaining()) {
40 			switch (state) {
41 			case State.PREPARE: {
42 				state = State.LAST_STREAM_ID;
43 				length = getBodyLength();
44 				break;
45 			}
46 			case State.LAST_STREAM_ID: {
47 				if (buffer.remaining() >= 4) {
48 					lastStreamId = buffer.get!int();
49 					lastStreamId &= 0x7F_FF_FF_FF;
50 					state = State.ERROR;
51 					length -= 4;
52 					if (length <= 0)
53 						return connectionFailure(buffer, cast(int)ErrorCode.FRAME_SIZE_ERROR, "invalid_go_away_frame");
54 				} else {
55 					state = State.LAST_STREAM_ID_BYTES;
56 					cursor = 4;
57 				}
58 				break;
59 			}
60 			case State.LAST_STREAM_ID_BYTES: {
61 				int currByte = buffer.get() & 0xFF;
62 				--cursor;
63 				lastStreamId += currByte << (8 * cursor);
64 				--length;
65 				if (cursor > 0 && length <= 0)
66 					return connectionFailure(buffer, cast(int)ErrorCode.FRAME_SIZE_ERROR, "invalid_go_away_frame");
67 				if (cursor == 0) {
68 					lastStreamId &= 0x7F_FF_FF_FF;
69 					state = State.ERROR;
70 					if (length == 0)
71 						return connectionFailure(buffer, cast(int)ErrorCode.FRAME_SIZE_ERROR, "invalid_go_away_frame");
72 				}
73 				break;
74 			}
75 			case State.ERROR: {
76 				if (buffer.remaining() >= 4) {
77 					error = buffer.get!int();
78 					state = State.PAYLOAD;
79 					length -= 4;
80 					if (length < 0)
81 						return connectionFailure(buffer, cast(int)ErrorCode.FRAME_SIZE_ERROR, "invalid_go_away_frame");
82 					if (length == 0)
83 						return onGoAway(lastStreamId, error, null);
84 				} else {
85 					state = State.ERROR_BYTES;
86 					cursor = 4;
87 				}
88 				break;
89 			}
90 			case State.ERROR_BYTES: {
91 				int currByte = buffer.get() & 0xFF;
92 				--cursor;
93 				error += currByte << (8 * cursor);
94 				--length;
95 				if (cursor > 0 && length <= 0)
96 					return connectionFailure(buffer, cast(int)ErrorCode.FRAME_SIZE_ERROR, "invalid_go_away_frame");
97 				if (cursor == 0) {
98 					state = State.PAYLOAD;
99 					if (length == 0)
100 						return onGoAway(lastStreamId, error, null);
101 				}
102 				break;
103 			}
104 			case State.PAYLOAD: {
105 				payload = new byte[length];
106 				if (buffer.remaining() >= length) {
107 					buffer.get(payload);
108 					return onGoAway(lastStreamId, error, payload);
109 				} else {
110 					state = State.PAYLOAD_BYTES;
111 					cursor = length;
112 				}
113 				break;
114 			}
115 			case State.PAYLOAD_BYTES: {
116 				payload[payload.length - cursor] = buffer.get();
117 				--cursor;
118 				if (cursor == 0)
119 					return onGoAway(lastStreamId, error, payload);
120 				break;
121 			}
122 			default: {
123 				throw new IllegalStateException("");
124 			}
125 			}
126 		}
127 		return false;
128 	}
129 
130 	private bool onGoAway(int lastStreamId, int error, byte[] payload) {
131 		GoAwayFrame frame = new GoAwayFrame(lastStreamId, error, payload);
132 		reset();
133 		notifyGoAway(frame);
134 		return true;
135 	}
136 
137 	private enum State {
138 		PREPARE, LAST_STREAM_ID, LAST_STREAM_ID_BYTES, ERROR, ERROR_BYTES, PAYLOAD, PAYLOAD_BYTES
139 	}
140 }