1
2
3
4
5
6
7
8
9
10
11
12
13
14
15 package io.netty5.handler.codec.http;
16
17 import io.netty5.buffer.api.BufferAllocator;
18 import io.netty5.channel.internal.DelegatingChannelHandlerContext;
19 import io.netty5.util.Resource;
20 import io.netty5.util.Send;
21 import io.netty5.channel.ChannelFutureListeners;
22 import io.netty5.channel.ChannelHandlerContext;
23 import io.netty5.util.concurrent.Future;
24
25 import java.util.ArrayList;
26 import java.util.Collection;
27 import java.util.List;
28
29 import static io.netty5.handler.codec.http.HttpResponseStatus.SWITCHING_PROTOCOLS;
30 import static io.netty5.handler.codec.http.HttpVersion.HTTP_1_1;
31 import static io.netty5.util.AsciiString.containsAllContentEqualsIgnoreCase;
32 import static io.netty5.util.AsciiString.containsContentEqualsIgnoreCase;
33 import static io.netty5.util.internal.StringUtil.COMMA;
34 import static java.util.Objects.requireNonNull;
35
36
37
38
39
40
41 public class HttpServerUpgradeHandler<C extends HttpContent<C>> extends HttpObjectAggregator<C> {
42
43
44
45
46 public interface SourceCodec {
47
48
49
50 void upgradeFrom(ChannelHandlerContext ctx);
51 }
52
53
54
55
56 public interface UpgradeCodec {
57
58
59
60
61 Collection<CharSequence> requiredUpgradeHeaders();
62
63
64
65
66
67
68
69
70
71 boolean prepareUpgradeResponse(ChannelHandlerContext ctx, FullHttpRequest upgradeRequest,
72 HttpHeaders upgradeHeaders);
73
74
75
76
77
78
79
80
81 void upgradeTo(ChannelHandlerContext ctx, FullHttpRequest upgradeRequest);
82 }
83
84
85
86
87 public interface UpgradeCodecFactory {
88
89
90
91
92
93
94
95 UpgradeCodec newUpgradeCodec(CharSequence protocol);
96 }
97
98
99
100
101
102
103 public static final class UpgradeEvent implements Resource<UpgradeEvent> {
104 private final CharSequence protocol;
105 private final FullHttpRequest upgradeRequest;
106
107 UpgradeEvent(CharSequence protocol, FullHttpRequest upgradeRequest) {
108 this.protocol = protocol;
109 this.upgradeRequest = upgradeRequest;
110 }
111
112
113
114
115 public CharSequence protocol() {
116 return protocol;
117 }
118
119
120
121
122 public FullHttpRequest upgradeRequest() {
123 return upgradeRequest;
124 }
125
126 @Override
127 public Send<UpgradeEvent> send() {
128 return upgradeRequest.send().map(UpgradeEvent.class, req -> new UpgradeEvent(protocol, req));
129 }
130
131 public UpgradeEvent copy() {
132 return new UpgradeEvent(protocol, upgradeRequest.copy());
133 }
134
135 @Override
136 public void close() {
137 upgradeRequest.close();
138 }
139
140 @Override
141 public boolean isAccessible() {
142 return upgradeRequest.isAccessible();
143 }
144
145 @Override
146 public UpgradeEvent touch(Object hint) {
147 upgradeRequest.touch(hint);
148 return this;
149 }
150
151 @Override
152 public String toString() {
153 return "UpgradeEvent [protocol=" + protocol + ", upgradeRequest=" + upgradeRequest + ']';
154 }
155 }
156
157 private final SourceCodec sourceCodec;
158 private final UpgradeCodecFactory upgradeCodecFactory;
159 private final boolean validateHeaders;
160 private boolean handlingUpgrade;
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176 public HttpServerUpgradeHandler(SourceCodec sourceCodec, UpgradeCodecFactory upgradeCodecFactory) {
177 this(sourceCodec, upgradeCodecFactory, 0);
178 }
179
180
181
182
183
184
185
186
187
188 public HttpServerUpgradeHandler(
189 SourceCodec sourceCodec, UpgradeCodecFactory upgradeCodecFactory, int maxContentLength) {
190 this(sourceCodec, upgradeCodecFactory, maxContentLength, true);
191 }
192
193
194
195
196
197
198
199
200
201
202 public HttpServerUpgradeHandler(SourceCodec sourceCodec, UpgradeCodecFactory upgradeCodecFactory,
203 int maxContentLength, boolean validateHeaders) {
204 super(maxContentLength);
205
206 this.sourceCodec = requireNonNull(sourceCodec, "sourceCodec");
207 this.upgradeCodecFactory = requireNonNull(upgradeCodecFactory, "upgradeCodecFactory");
208 this.validateHeaders = validateHeaders;
209 }
210
211 @Override
212 protected void decodeAndClose(final ChannelHandlerContext ctx, HttpObject msg)
213 throws Exception {
214
215 if (!handlingUpgrade) {
216
217 if (msg instanceof HttpRequest) {
218 HttpRequest req = (HttpRequest) msg;
219 if (req.headers().contains(HttpHeaderNames.UPGRADE) &&
220 shouldHandleUpgradeRequest(req)) {
221 handlingUpgrade = true;
222 } else {
223 ctx.fireChannelRead(msg);
224 return;
225 }
226 } else {
227 ctx.fireChannelRead(msg);
228 return;
229 }
230 }
231
232 FullHttpRequest fullRequest;
233 if (msg instanceof FullHttpRequest) {
234 fullRequest = (FullHttpRequest) msg;
235 tryUpgrade(ctx, fullRequest);
236 } else {
237
238 super.decodeAndClose(new DelegatingChannelHandlerContext(ctx) {
239 @Override
240 public ChannelHandlerContext fireChannelRead(Object msg) {
241
242 handlingUpgrade = false;
243 tryUpgrade(ctx, (FullHttpRequest) msg);
244 return this;
245 }
246 }, msg);
247 }
248 }
249
250 private void tryUpgrade(ChannelHandlerContext ctx, FullHttpRequest request) {
251 if (!upgrade(ctx, request)) {
252
253
254
255 ctx.fireChannelRead(request);
256 }
257 }
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272 protected boolean shouldHandleUpgradeRequest(HttpRequest req) {
273 return true;
274 }
275
276
277
278
279
280
281
282
283
284 private boolean upgrade(final ChannelHandlerContext ctx, final FullHttpRequest request) {
285
286 final List<CharSequence> requestedProtocols = splitHeader(request.headers().get(HttpHeaderNames.UPGRADE));
287 final int numRequestedProtocols = requestedProtocols.size();
288 UpgradeCodec upgradeCodec = null;
289 CharSequence upgradeProtocol = null;
290 for (int i = 0; i < numRequestedProtocols; i ++) {
291 final CharSequence p = requestedProtocols.get(i);
292 final UpgradeCodec c = upgradeCodecFactory.newUpgradeCodec(p);
293 if (c != null) {
294 upgradeProtocol = p;
295 upgradeCodec = c;
296 break;
297 }
298 }
299
300 if (upgradeCodec == null) {
301
302 return false;
303 }
304
305
306 List<String> connectionHeaderValues = request.headers().getAll(HttpHeaderNames.CONNECTION);
307
308 if (connectionHeaderValues == null || connectionHeaderValues.isEmpty()) {
309 return false;
310 }
311
312 final StringBuilder concatenatedConnectionValue = new StringBuilder(connectionHeaderValues.size() * 10);
313 for (CharSequence connectionHeaderValue : connectionHeaderValues) {
314 concatenatedConnectionValue.append(connectionHeaderValue).append(COMMA);
315 }
316 concatenatedConnectionValue.setLength(concatenatedConnectionValue.length() - 1);
317
318
319 Collection<CharSequence> requiredHeaders = upgradeCodec.requiredUpgradeHeaders();
320 List<CharSequence> values = splitHeader(concatenatedConnectionValue);
321 if (!containsContentEqualsIgnoreCase(values, HttpHeaderNames.UPGRADE) ||
322 !containsAllContentEqualsIgnoreCase(values, requiredHeaders)) {
323 return false;
324 }
325
326
327 for (CharSequence requiredHeader : requiredHeaders) {
328 if (!request.headers().contains(requiredHeader)) {
329 return false;
330 }
331 }
332
333
334
335 final FullHttpResponse upgradeResponse = createUpgradeResponse(ctx.bufferAllocator(), upgradeProtocol);
336 if (!upgradeCodec.prepareUpgradeResponse(ctx, request, upgradeResponse.headers())) {
337 return false;
338 }
339
340
341
342
343
344 Future<Void> writeComplete = ctx.writeAndFlush(upgradeResponse);
345
346 sourceCodec.upgradeFrom(ctx);
347 upgradeCodec.upgradeTo(ctx, request);
348
349
350 ctx.fireChannelInboundEvent(new UpgradeEvent(upgradeProtocol, request));
351
352
353 ctx.pipeline().remove(HttpServerUpgradeHandler.this);
354
355
356
357
358 writeComplete.addListener(ctx.channel(), ChannelFutureListeners.CLOSE_ON_FAILURE);
359 return true;
360 }
361
362
363
364
365 private FullHttpResponse createUpgradeResponse(BufferAllocator allocator, CharSequence upgradeProtocol) {
366 DefaultFullHttpResponse res = new DefaultFullHttpResponse(
367 HTTP_1_1, SWITCHING_PROTOCOLS, allocator.allocate(0), validateHeaders);
368 res.headers().add(HttpHeaderNames.CONNECTION, HttpHeaderValues.UPGRADE);
369 res.headers().add(HttpHeaderNames.UPGRADE, upgradeProtocol);
370 return res;
371 }
372
373
374
375
376
377 private static List<CharSequence> splitHeader(CharSequence header) {
378 final StringBuilder builder = new StringBuilder(header.length());
379 final List<CharSequence> protocols = new ArrayList<>(4);
380 for (int i = 0; i < header.length(); ++i) {
381 char c = header.charAt(i);
382 if (Character.isWhitespace(c)) {
383
384 continue;
385 }
386 if (c == ',') {
387
388 protocols.add(builder.toString());
389 builder.setLength(0);
390 } else {
391 builder.append(c);
392 }
393 }
394
395
396 if (builder.length() > 0) {
397 protocols.add(builder.toString());
398 }
399
400 return protocols;
401 }
402 }