View Javadoc
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.http.websocketx.extensions;
17  
18  import static io.netty.util.internal.ObjectUtil.checkNonEmpty;
19  
20  import io.netty.buffer.ByteBuf;
21  import io.netty.buffer.Unpooled;
22  import io.netty.channel.ChannelDuplexHandler;
23  import io.netty.channel.ChannelFuture;
24  import io.netty.channel.ChannelFutureListener;
25  import io.netty.channel.ChannelHandlerContext;
26  import io.netty.channel.ChannelPromise;
27  import io.netty.handler.codec.http.DefaultHttpRequest;
28  import io.netty.handler.codec.http.DefaultHttpResponse;
29  import io.netty.handler.codec.http.HttpHeaderNames;
30  import io.netty.handler.codec.http.HttpHeaders;
31  import io.netty.handler.codec.http.HttpRequest;
32  import io.netty.handler.codec.http.HttpResponse;
33  import io.netty.handler.codec.http.HttpResponseStatus;
34  import io.netty.handler.codec.http.LastHttpContent;
35  import io.netty.util.ReferenceCountUtil;
36  import io.netty.util.internal.ObjectUtil;
37  
38  import java.util.ArrayDeque;
39  import java.util.ArrayList;
40  import java.util.Arrays;
41  import java.util.Collections;
42  import java.util.Iterator;
43  import java.util.List;
44  import java.util.Queue;
45  
46  /**
47   * This handler negotiates and initializes the WebSocket Extensions.
48   *
49   * It negotiates the extensions based on the client desired order,
50   * ensures that the successfully negotiated extensions are consistent between them,
51   * and initializes the channel pipeline with the extension decoder and encoder.
52   *
53   * Find a basic implementation for compression extensions at
54   * <tt>io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler</tt>.
55   */
56  public class WebSocketServerExtensionHandler extends ChannelDuplexHandler {
57      private static final int DEFAULT_MAX_PIPELINE_DEPTH = 128;
58      private final int maxPipelineDepth;
59      private final List<WebSocketServerExtensionHandshaker> extensionHandshakers;
60  
61      private final Queue<List<WebSocketServerExtension>> validExtensions =
62              new ArrayDeque<List<WebSocketServerExtension>>(4);
63  
64      /**
65       * Constructor
66       *
67       * @param extensionHandshakers
68       *      The extension handshaker in priority order. A handshaker could be repeated many times
69       *      with fallback configuration.
70       */
71      public WebSocketServerExtensionHandler(WebSocketServerExtensionHandshaker... extensionHandshakers) {
72          this(DEFAULT_MAX_PIPELINE_DEPTH, extensionHandshakers);
73      }
74  
75      /**
76       * Constructor
77       *
78       * @param maxPipelineDepth
79       *      The maximum number of pipelined upgrade requests.
80       * @param extensionHandshakers
81       *      The extension handshaker in priority order. A handshaker could be repeated many times
82       *      with fallback configuration.
83       */
84      public WebSocketServerExtensionHandler(
85          int maxPipelineDepth, WebSocketServerExtensionHandshaker... extensionHandshakers) {
86          this.maxPipelineDepth = ObjectUtil.checkPositive(maxPipelineDepth, "maxPipelineDepth");
87          this.extensionHandshakers = Arrays.asList(checkNonEmpty(extensionHandshakers, "extensionHandshakers"));
88      }
89  
90      @Override
91      public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
92          // JDK type checks vs non-implemented interfaces costs O(N), where
93          // N is the number of interfaces already implemented by the concrete type that's being tested.
94          // The only requirement for this call is to make HttpRequest(s) implementors to call onHttpRequestChannelRead
95          // and super.channelRead the others, but due to the O(n) cost we perform few fast-path for commonly met
96          // singleton and/or concrete types, to save performing such slow type checks.
97          if (msg != LastHttpContent.EMPTY_LAST_CONTENT) {
98              if (msg instanceof DefaultHttpRequest) {
99                  // fast-path
100                 onHttpRequestChannelRead(ctx, (DefaultHttpRequest) msg);
101             } else if (msg instanceof HttpRequest) {
102                 // slow path
103                 onHttpRequestChannelRead(ctx, (HttpRequest) msg);
104             } else {
105                 super.channelRead(ctx, msg);
106             }
107         } else {
108             super.channelRead(ctx, msg);
109         }
110     }
111 
112     /**
113      * This is a method exposed to perform fail-fast checks of user-defined http types.<p>
114      * eg:<br>
115      * If the user has defined a specific {@link HttpRequest} type i.e.{@code CustomHttpRequest} and
116      * {@link #channelRead} can receive {@link LastHttpContent#EMPTY_LAST_CONTENT} {@code msg}
117      * types too, can override it like this:
118      * <pre>
119      *     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
120      *         if (msg != LastHttpContent.EMPTY_LAST_CONTENT) {
121      *             if (msg instanceof CustomHttpRequest) {
122      *                 onHttpRequestChannelRead(ctx, (CustomHttpRequest) msg);
123      *             } else {
124      *                 // if it's handling other HttpRequest types it MUST use onHttpRequestChannelRead again
125      *                 // or have to delegate it to super.channelRead (that can perform redundant checks).
126      *                 // If msg is not implementing HttpRequest, it can call ctx.fireChannelRead(msg) on it
127      *                 // ...
128      *                 super.channelRead(ctx, msg);
129      *             }
130      *         } else {
131      *             // given that msg isn't a HttpRequest type we can just skip calling super.channelRead
132      *             ctx.fireChannelRead(msg);
133      *         }
134      *     }
135      * </pre>
136      * <strong>IMPORTANT:</strong>
137      * It already call {@code super.channelRead(ctx, request)} before returning.
138      */
139     protected void onHttpRequestChannelRead(ChannelHandlerContext ctx, HttpRequest request) throws Exception {
140         if (maxPipelineDepth <= validExtensions.size()) {
141             ReferenceCountUtil.release(request);
142             ctx.close();
143             throw new IllegalStateException("maxPipelineDepth exceeded: " + maxPipelineDepth);
144         }
145 
146         List<WebSocketServerExtension> validExtensionsList = null;
147 
148         if (WebSocketExtensionUtil.isWebsocketUpgrade(request.headers())) {
149             String extensionsHeader = request.headers().getAsString(HttpHeaderNames.SEC_WEBSOCKET_EXTENSIONS);
150 
151             if (extensionsHeader != null) {
152                 List<WebSocketExtensionData> extensions =
153                         WebSocketExtensionUtil.extractExtensions(extensionsHeader);
154                 int rsv = 0;
155 
156                 for (WebSocketExtensionData extensionData : extensions) {
157                     Iterator<WebSocketServerExtensionHandshaker> extensionHandshakersIterator =
158                             extensionHandshakers.iterator();
159                     WebSocketServerExtension validExtension = null;
160 
161                     while (validExtension == null && extensionHandshakersIterator.hasNext()) {
162                         WebSocketServerExtensionHandshaker extensionHandshaker =
163                                 extensionHandshakersIterator.next();
164                         validExtension = extensionHandshaker.handshakeExtension(extensionData);
165                     }
166 
167                     if (validExtension != null && ((validExtension.rsv() & rsv) == 0)) {
168                         if (validExtensionsList == null) {
169                             validExtensionsList = new ArrayList<WebSocketServerExtension>(1);
170                         }
171                         rsv = rsv | validExtension.rsv();
172                         validExtensionsList.add(validExtension);
173                     }
174                 }
175             }
176         }
177 
178         if (validExtensionsList == null) {
179             validExtensionsList = Collections.emptyList();
180         }
181         validExtensions.offer(validExtensionsList);
182 
183         super.channelRead(ctx, request);
184     }
185 
186     @Override
187     public void write(final ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
188         if (msg != Unpooled.EMPTY_BUFFER && !(msg instanceof ByteBuf)) {
189             if (msg instanceof DefaultHttpResponse) {
190                 onHttpResponseWrite(ctx, (DefaultHttpResponse) msg, promise);
191             } else if (msg instanceof HttpResponse) {
192                 onHttpResponseWrite(ctx, (HttpResponse) msg, promise);
193             } else {
194                 super.write(ctx, msg, promise);
195             }
196         } else {
197             super.write(ctx, msg, promise);
198         }
199     }
200 
201     /**
202      * This is a method exposed to perform fail-fast checks of user-defined http types.<p>
203      * eg:<br>
204      * If the user has defined a specific {@link HttpResponse} type i.e.{@code CustomHttpResponse} and
205      * {@link #write} can receive {@link ByteBuf} {@code msg} types too, it can be overridden like this:
206      * <pre>
207      *     public void write(final ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
208      *         if (msg != Unpooled.EMPTY_BUFFER && !(msg instanceof ByteBuf)) {
209      *             if (msg instanceof CustomHttpResponse) {
210      *                 onHttpResponseWrite(ctx, (CustomHttpResponse) msg, promise);
211      *             } else {
212      *                 // if it's handling other HttpResponse types it MUST use onHttpResponseWrite again
213      *                 // or have to delegate it to super.write (that can perform redundant checks).
214      *                 // If msg is not implementing HttpResponse, it can call ctx.write(msg, promise) on it
215      *                 // ...
216      *                 super.write(ctx, msg, promise);
217      *             }
218      *         } else {
219      *             // given that msg isn't a HttpResponse type we can just skip calling super.write
220      *             ctx.write(msg, promise);
221      *         }
222      *     }
223      * </pre>
224      * <strong>IMPORTANT:</strong>
225      * It already call {@code super.write(ctx, response, promise)} before returning.
226      */
227     protected void onHttpResponseWrite(ChannelHandlerContext ctx, HttpResponse response, ChannelPromise promise)
228             throws Exception {
229         List<WebSocketServerExtension> validExtensionsList = validExtensions.poll();
230         // checking the status is faster than looking at headers so we do this first
231         if (HttpResponseStatus.SWITCHING_PROTOCOLS.equals(response.status())) {
232             handlePotentialUpgrade(ctx, promise, response, validExtensionsList);
233         }
234         super.write(ctx, response, promise);
235     }
236 
237     private void handlePotentialUpgrade(final ChannelHandlerContext ctx,
238                                         ChannelPromise promise, HttpResponse httpResponse,
239                                         final List<WebSocketServerExtension> validExtensionsList) {
240         HttpHeaders headers = httpResponse.headers();
241 
242         if (WebSocketExtensionUtil.isWebsocketUpgrade(headers)) {
243             if (validExtensionsList != null && !validExtensionsList.isEmpty()) {
244                 String headerValue = headers.getAsString(HttpHeaderNames.SEC_WEBSOCKET_EXTENSIONS);
245                 List<WebSocketExtensionData> extraExtensions =
246                   new ArrayList<WebSocketExtensionData>(extensionHandshakers.size());
247                 for (WebSocketServerExtension extension : validExtensionsList) {
248                     extraExtensions.add(extension.newReponseData());
249                 }
250                 String newHeaderValue = WebSocketExtensionUtil
251                   .computeMergeExtensionsHeaderValue(headerValue, extraExtensions);
252                 promise.addListener(new ChannelFutureListener() {
253                     @Override
254                     public void operationComplete(ChannelFuture future) {
255                         if (future.isSuccess()) {
256                             for (WebSocketServerExtension extension : validExtensionsList) {
257                                 WebSocketExtensionDecoder decoder = extension.newExtensionDecoder();
258                                 WebSocketExtensionEncoder encoder = extension.newExtensionEncoder();
259                                 String name = ctx.name();
260                                 ctx.pipeline()
261                                     .addAfter(name, decoder.getClass().getName(), decoder)
262                                     .addAfter(name, encoder.getClass().getName(), encoder);
263                             }
264                         }
265                     }
266                 });
267 
268                 if (newHeaderValue != null) {
269                     headers.set(HttpHeaderNames.SEC_WEBSOCKET_EXTENSIONS, newHeaderValue);
270                 }
271             }
272 
273             promise.addListener(new ChannelFutureListener() {
274                 @Override
275                 public void operationComplete(ChannelFuture future) {
276                     if (future.isSuccess()) {
277                         ctx.pipeline().remove(WebSocketServerExtensionHandler.this);
278                     }
279                 }
280             });
281         }
282     }
283 }