1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.handler.codec.http.websocketx;
17
18 import io.netty.channel.ChannelFuture;
19 import io.netty.channel.ChannelHandlerContext;
20 import io.netty.channel.ChannelInboundHandlerAdapter;
21 import io.netty.channel.ChannelPipeline;
22 import io.netty.channel.ChannelPromise;
23 import io.netty.handler.codec.http.HttpHeaderNames;
24 import io.netty.handler.codec.http.HttpObject;
25 import io.netty.handler.codec.http.HttpRequest;
26 import io.netty.handler.codec.http.HttpResponse;
27 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler.ServerHandshakeStateEvent;
28 import io.netty.handler.ssl.SslHandler;
29 import io.netty.util.ReferenceCountUtil;
30 import io.netty.util.concurrent.Future;
31
32 import java.util.concurrent.TimeUnit;
33
34 import static io.netty.handler.codec.http.HttpUtil.*;
35 import static io.netty.util.internal.ObjectUtil.*;
36
37
38
39
40 class WebSocketServerProtocolHandshakeHandler extends ChannelInboundHandlerAdapter {
41
42 private final WebSocketServerProtocolConfig serverConfig;
43 private ChannelHandlerContext ctx;
44 private ChannelPromise handshakePromise;
45 private boolean isWebSocketPath;
46
47 WebSocketServerProtocolHandshakeHandler(WebSocketServerProtocolConfig serverConfig) {
48 this.serverConfig = checkNotNull(serverConfig, "serverConfig");
49 }
50
51 @Override
52 public void handlerAdded(ChannelHandlerContext ctx) {
53 this.ctx = ctx;
54 handshakePromise = ctx.newPromise();
55 }
56
57 @Override
58 public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
59 final HttpObject httpObject = (HttpObject) msg;
60
61 if (httpObject instanceof HttpRequest) {
62 final HttpRequest req = (HttpRequest) httpObject;
63 isWebSocketPath = isWebSocketPath(req);
64 if (!isWebSocketPath) {
65 ctx.fireChannelRead(msg);
66 return;
67 }
68
69 try {
70 final WebSocketServerHandshaker handshaker = WebSocketServerHandshakerFactory.resolveHandshaker(
71 req,
72 getWebSocketLocation(ctx.pipeline(), req, serverConfig.websocketPath()),
73 serverConfig.subprotocols(), serverConfig.decoderConfig());
74 final ChannelPromise localHandshakePromise = handshakePromise;
75 if (handshaker == null) {
76 WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
77 } else {
78
79
80
81
82
83 WebSocketServerProtocolHandler.setHandshaker(ctx.channel(), handshaker);
84 ctx.pipeline().remove(this);
85
86
87 final ChannelFuture handshakeFuture = handshaker.handshake(ctx, req);
88 handshakeFuture.addListener(future -> {
89 if (!future.isSuccess()) {
90 localHandshakePromise.tryFailure(future.cause());
91 ctx.fireExceptionCaught(future.cause());
92 } else {
93 localHandshakePromise.trySuccess();
94
95 ctx.fireUserEventTriggered(
96 ServerHandshakeStateEvent.HANDSHAKE_COMPLETE);
97 ctx.fireUserEventTriggered(
98 new WebSocketServerProtocolHandler.HandshakeComplete(
99 req.uri(), req.headers(), handshaker.selectedSubprotocol()));
100 }
101 });
102 applyHandshakeTimeout();
103 }
104 } finally {
105 ReferenceCountUtil.release(req);
106 }
107 } else if (!isWebSocketPath) {
108 ctx.fireChannelRead(msg);
109 } else {
110 ReferenceCountUtil.release(msg);
111 }
112 }
113
114 private boolean isWebSocketPath(HttpRequest req) {
115 String websocketPath = serverConfig.websocketPath();
116 String uri = req.uri();
117 return serverConfig.checkStartsWith()
118 ? uri.startsWith(websocketPath) && ("/".equals(websocketPath) || checkNextUri(uri, websocketPath))
119 : uri.equals(websocketPath);
120 }
121
122 private boolean checkNextUri(String uri, String websocketPath) {
123 int len = websocketPath.length();
124 if (uri.length() > len) {
125 char nextUri = uri.charAt(len);
126 return nextUri == '/' || nextUri == '?';
127 }
128 return true;
129 }
130
131 private static String getWebSocketLocation(ChannelPipeline cp, HttpRequest req, String path) {
132 String protocol = "ws";
133 if (cp.get(SslHandler.class) != null) {
134
135 protocol = "wss";
136 }
137 String host = req.headers().get(HttpHeaderNames.HOST);
138 return protocol + "://" + host + path;
139 }
140
141 private void applyHandshakeTimeout() {
142 final ChannelPromise localHandshakePromise = handshakePromise;
143 final long handshakeTimeoutMillis = serverConfig.handshakeTimeoutMillis();
144 if (handshakeTimeoutMillis <= 0 || localHandshakePromise.isDone()) {
145 return;
146 }
147
148 final Future<?> timeoutFuture = ctx.executor().schedule(new Runnable() {
149 @Override
150 public void run() {
151 if (!localHandshakePromise.isDone() &&
152 localHandshakePromise.tryFailure(new WebSocketServerHandshakeException("handshake timed out"))) {
153 ctx.flush()
154 .fireUserEventTriggered(ServerHandshakeStateEvent.HANDSHAKE_TIMEOUT)
155 .close();
156 }
157 }
158 }, handshakeTimeoutMillis, TimeUnit.MILLISECONDS);
159
160
161 localHandshakePromise.addListener(f -> timeoutFuture.cancel(false));
162 }
163 }