View Javadoc
1   /*
2    * Copyright 2013 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    *   http://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 org.jboss.netty.handler.codec.spdy;
17  
18  import org.jboss.netty.channel.Channel;
19  import org.jboss.netty.channel.ChannelDownstreamHandler;
20  import org.jboss.netty.channel.ChannelEvent;
21  import org.jboss.netty.channel.ChannelFuture;
22  import org.jboss.netty.channel.ChannelFutureListener;
23  import org.jboss.netty.channel.ChannelHandlerContext;
24  import org.jboss.netty.channel.ChannelStateEvent;
25  import org.jboss.netty.channel.Channels;
26  import org.jboss.netty.channel.ExceptionEvent;
27  import org.jboss.netty.channel.MessageEvent;
28  import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
29  
30  import java.net.SocketAddress;
31  import java.nio.channels.ClosedChannelException;
32  import java.util.concurrent.atomic.AtomicInteger;
33  
34  import static org.jboss.netty.handler.codec.spdy.SpdyCodecUtil.*;
35  
36  /**
37   * Manages streams within a SPDY session.
38   */
39  public class SpdySessionHandler extends SimpleChannelUpstreamHandler
40          implements ChannelDownstreamHandler {
41  
42      private static final SpdyProtocolException PROTOCOL_EXCEPTION = new SpdyProtocolException();
43  
44      private static final int DEFAULT_WINDOW_SIZE = 64 * 1024; // 64 KB default initial window size
45      private volatile int initialSendWindowSize    = DEFAULT_WINDOW_SIZE;
46      private volatile int initialReceiveWindowSize = DEFAULT_WINDOW_SIZE;
47      private volatile int initialSessionReceiveWindowSize = DEFAULT_WINDOW_SIZE;
48  
49      private final SpdySession spdySession = new SpdySession(initialSendWindowSize, initialReceiveWindowSize);
50      private volatile int lastGoodStreamId;
51  
52      private static final int DEFAULT_MAX_CONCURRENT_STREAMS = Integer.MAX_VALUE;
53      private volatile int remoteConcurrentStreams = DEFAULT_MAX_CONCURRENT_STREAMS;
54      private volatile int localConcurrentStreams  = DEFAULT_MAX_CONCURRENT_STREAMS;
55  
56      private final Object flowControlLock = new Object();
57  
58      private final AtomicInteger pings = new AtomicInteger();
59  
60      private volatile boolean sentGoAwayFrame;
61      private volatile boolean receivedGoAwayFrame;
62  
63      private volatile ChannelFutureListener closeSessionFutureListener;
64  
65      private final boolean server;
66      private final int minorVersion;
67  
68      /**
69       * Creates a new session handler.
70       *
71       * @param spdyVersion the protocol version
72       * @param server      {@code true} if and only if this session handler should
73       *                    handle the server endpoint of the connection.
74       *                    {@code false} if and only if this session handler should
75       *                    handle the client endpoint of the connection.
76       */
77      public SpdySessionHandler(SpdyVersion spdyVersion, boolean server) {
78          if (spdyVersion == null) {
79              throw new NullPointerException("spdyVersion");
80          }
81          this.server = server;
82          minorVersion = spdyVersion.getMinorVersion();
83      }
84  
85      public void setSessionReceiveWindowSize(int sessionReceiveWindowSize) {
86        if (sessionReceiveWindowSize < 0) {
87          throw new IllegalArgumentException("sessionReceiveWindowSize");
88        }
89        // This will not send a window update frame immediately.
90        // If this value increases the allowed receive window size,
91        // a WINDOW_UPDATE frame will be sent when only half of the
92        // session window size remains during data frame processing.
93        // If this value decreases the allowed receive window size,
94        // the window will be reduced as data frames are processed.
95        initialSessionReceiveWindowSize = sessionReceiveWindowSize;
96      }
97  
98      @Override
99      public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
100             throws Exception {
101 
102         Object msg = e.getMessage();
103         if (msg instanceof SpdyDataFrame) {
104 
105             /*
106              * SPDY Data frame processing requirements:
107              *
108              * If an endpoint receives a data frame for a Stream-ID which is not open
109              * and the endpoint has not sent a GOAWAY frame, it must issue a stream error
110              * with the error code INVALID_STREAM for the Stream-ID.
111              *
112              * If an endpoint which created the stream receives a data frame before receiving
113              * a SYN_REPLY on that stream, it is a protocol error, and the recipient must
114              * issue a stream error with the status code PROTOCOL_ERROR for the Stream-ID.
115              *
116              * If an endpoint receives multiple data frames for invalid Stream-IDs,
117              * it may close the session.
118              *
119              * If an endpoint refuses a stream it must ignore any data frames for that stream.
120              *
121              * If an endpoint receives a data frame after the stream is half-closed from the
122              * sender, it must send a RST_STREAM frame with the status STREAM_ALREADY_CLOSED.
123              *
124              * If an endpoint receives a data frame after the stream is closed, it must send
125              * a RST_STREAM frame with the status PROTOCOL_ERROR.
126              */
127 
128             SpdyDataFrame spdyDataFrame = (SpdyDataFrame) msg;
129             int streamId = spdyDataFrame.getStreamId();
130 
131             int deltaWindowSize = -1 * spdyDataFrame.getData().readableBytes();
132             int newSessionWindowSize =
133                 spdySession.updateReceiveWindowSize(SPDY_SESSION_STREAM_ID, deltaWindowSize);
134 
135             // Check if session window size is reduced beyond allowable lower bound
136             if (newSessionWindowSize < 0) {
137                 issueSessionError(ctx, e.getChannel(), e.getRemoteAddress(), SpdySessionStatus.PROTOCOL_ERROR);
138                 return;
139             }
140 
141             // Send a WINDOW_UPDATE frame if less than half the session window size remains
142             if (newSessionWindowSize <= initialSessionReceiveWindowSize / 2) {
143                 int sessionDeltaWindowSize = initialSessionReceiveWindowSize - newSessionWindowSize;
144                 spdySession.updateReceiveWindowSize(SPDY_SESSION_STREAM_ID, sessionDeltaWindowSize);
145                 SpdyWindowUpdateFrame spdyWindowUpdateFrame =
146                     new DefaultSpdyWindowUpdateFrame(SPDY_SESSION_STREAM_ID, sessionDeltaWindowSize);
147                 Channels.write(
148                         ctx, Channels.future(e.getChannel()), spdyWindowUpdateFrame, e.getRemoteAddress());
149             }
150 
151             // Check if we received a data frame for a Stream-ID which is not open
152             if (!spdySession.isActiveStream(streamId)) {
153                 if (streamId <= lastGoodStreamId) {
154                     issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.PROTOCOL_ERROR);
155                 } else if (!sentGoAwayFrame) {
156                     issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.INVALID_STREAM);
157                 }
158                 return;
159             }
160 
161             // Check if we received a data frame for a stream which is half-closed
162             if (spdySession.isRemoteSideClosed(streamId)) {
163                 issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.STREAM_ALREADY_CLOSED);
164                 return;
165             }
166 
167             // Check if we received a data frame before receiving a SYN_REPLY
168             if (!isRemoteInitiatedId(streamId) && !spdySession.hasReceivedReply(streamId)) {
169                 issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.PROTOCOL_ERROR);
170                 return;
171             }
172 
173             /*
174              * SPDY Data frame flow control processing requirements:
175              *
176              * Recipient should not send a WINDOW_UPDATE frame as it consumes the last data frame.
177              */
178 
179             // Update receive window size
180             int newWindowSize = spdySession.updateReceiveWindowSize(streamId, deltaWindowSize);
181 
182             // Window size can become negative if we sent a SETTINGS frame that reduces the
183             // size of the transfer window after the peer has written data frames.
184             // The value is bounded by the length that SETTINGS frame decrease the window.
185             // This difference is stored for the session when writing the SETTINGS frame
186             // and is cleared once we send a WINDOW_UPDATE frame.
187             if (newWindowSize < spdySession.getReceiveWindowSizeLowerBound(streamId)) {
188                 issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.FLOW_CONTROL_ERROR);
189                 return;
190             }
191 
192             // Window size became negative due to sender writing frame before receiving SETTINGS
193             // Send data frames upstream in initialReceiveWindowSize chunks
194             if (newWindowSize < 0) {
195                 while (spdyDataFrame.getData().readableBytes() > initialReceiveWindowSize) {
196                     SpdyDataFrame partialDataFrame = new DefaultSpdyDataFrame(streamId);
197                     partialDataFrame.setData(spdyDataFrame.getData().readSlice(initialReceiveWindowSize));
198                     Channels.fireMessageReceived(ctx, partialDataFrame, e.getRemoteAddress());
199                 }
200             }
201 
202             // Send a WINDOW_UPDATE frame if less than half the stream window size remains
203             if (newWindowSize <= initialReceiveWindowSize / 2 && !spdyDataFrame.isLast()) {
204                 int streamDeltaWindowSize = initialReceiveWindowSize - newWindowSize;
205                 spdySession.updateReceiveWindowSize(streamId, streamDeltaWindowSize);
206                 SpdyWindowUpdateFrame spdyWindowUpdateFrame =
207                         new DefaultSpdyWindowUpdateFrame(streamId, streamDeltaWindowSize);
208                 Channels.write(
209                         ctx, Channels.future(e.getChannel()), spdyWindowUpdateFrame, e.getRemoteAddress());
210             }
211 
212             // Close the remote side of the stream if this is the last frame
213             if (spdyDataFrame.isLast()) {
214                 halfCloseStream(streamId, true, e.getFuture());
215             }
216 
217         } else if (msg instanceof SpdySynStreamFrame) {
218 
219             /*
220              * SPDY SYN_STREAM frame processing requirements:
221              *
222              * If an endpoint receives a SYN_STREAM with a Stream-ID that is less than
223              * any previously received SYN_STREAM, it must issue a session error with
224              * the status PROTOCOL_ERROR.
225              *
226              * If an endpoint receives multiple SYN_STREAM frames with the same active
227              * Stream-ID, it must issue a stream error with the status code PROTOCOL_ERROR.
228              *
229              * The recipient can reject a stream by sending a stream error with the
230              * status code REFUSED_STREAM.
231              */
232 
233             SpdySynStreamFrame spdySynStreamFrame = (SpdySynStreamFrame) msg;
234             int streamId = spdySynStreamFrame.getStreamId();
235 
236             // Check if we received a valid SYN_STREAM frame
237             if (spdySynStreamFrame.isInvalid() ||
238                 !isRemoteInitiatedId(streamId) ||
239                 spdySession.isActiveStream(streamId)) {
240                 issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.PROTOCOL_ERROR);
241                 return;
242             }
243 
244             // Stream-IDs must be monotonically increasing
245             if (streamId <= lastGoodStreamId) {
246                 issueSessionError(ctx, e.getChannel(), e.getRemoteAddress(), SpdySessionStatus.PROTOCOL_ERROR);
247                 return;
248             }
249 
250             // Try to accept the stream
251             byte priority = spdySynStreamFrame.getPriority();
252             boolean remoteSideClosed = spdySynStreamFrame.isLast();
253             boolean localSideClosed = spdySynStreamFrame.isUnidirectional();
254             if (!acceptStream(streamId, priority, remoteSideClosed, localSideClosed)) {
255                 issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.REFUSED_STREAM);
256                 return;
257             }
258 
259         } else if (msg instanceof SpdySynReplyFrame) {
260 
261             /*
262              * SPDY SYN_REPLY frame processing requirements:
263              *
264              * If an endpoint receives multiple SYN_REPLY frames for the same active Stream-ID
265              * it must issue a stream error with the status code STREAM_IN_USE.
266              */
267 
268             SpdySynReplyFrame spdySynReplyFrame = (SpdySynReplyFrame) msg;
269             int streamId = spdySynReplyFrame.getStreamId();
270 
271             // Check if we received a valid SYN_REPLY frame
272             if (spdySynReplyFrame.isInvalid() ||
273                 isRemoteInitiatedId(streamId) ||
274                 spdySession.isRemoteSideClosed(streamId)) {
275                 issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.INVALID_STREAM);
276                 return;
277             }
278 
279             // Check if we have received multiple frames for the same Stream-ID
280             if (spdySession.hasReceivedReply(streamId)) {
281                 issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.STREAM_IN_USE);
282                 return;
283             }
284 
285             spdySession.receivedReply(streamId);
286 
287             // Close the remote side of the stream if this is the last frame
288             if (spdySynReplyFrame.isLast()) {
289                 halfCloseStream(streamId, true, e.getFuture());
290             }
291 
292         } else if (msg instanceof SpdyRstStreamFrame) {
293 
294             /*
295              * SPDY RST_STREAM frame processing requirements:
296              *
297              * After receiving a RST_STREAM on a stream, the receiver must not send
298              * additional frames on that stream.
299              *
300              * An endpoint must not send a RST_STREAM in response to a RST_STREAM.
301              */
302 
303             SpdyRstStreamFrame spdyRstStreamFrame = (SpdyRstStreamFrame) msg;
304             removeStream(spdyRstStreamFrame.getStreamId(), e.getFuture());
305 
306         } else if (msg instanceof SpdySettingsFrame) {
307 
308             SpdySettingsFrame spdySettingsFrame = (SpdySettingsFrame) msg;
309 
310             int settingsMinorVersion = spdySettingsFrame.getValue(SpdySettingsFrame.SETTINGS_MINOR_VERSION);
311             if (settingsMinorVersion >= 0 && settingsMinorVersion != minorVersion) {
312                 // Settings frame had the wrong minor version
313                 issueSessionError(ctx, e.getChannel(), e.getRemoteAddress(), SpdySessionStatus.PROTOCOL_ERROR);
314                 return;
315             }
316 
317             int newConcurrentStreams =
318                 spdySettingsFrame.getValue(SpdySettingsFrame.SETTINGS_MAX_CONCURRENT_STREAMS);
319             if (newConcurrentStreams >= 0) {
320                 remoteConcurrentStreams = newConcurrentStreams;
321             }
322 
323             // Persistence flag are inconsistent with the use of SETTINGS to communicate
324             // the initial window size. Remove flags from the sender requesting that the
325             // value be persisted. Remove values that the sender indicates are persisted.
326             if (spdySettingsFrame.isPersisted(SpdySettingsFrame.SETTINGS_INITIAL_WINDOW_SIZE)) {
327                 spdySettingsFrame.removeValue(SpdySettingsFrame.SETTINGS_INITIAL_WINDOW_SIZE);
328             }
329             spdySettingsFrame.setPersistValue(SpdySettingsFrame.SETTINGS_INITIAL_WINDOW_SIZE, false);
330 
331             int newInitialWindowSize =
332                 spdySettingsFrame.getValue(SpdySettingsFrame.SETTINGS_INITIAL_WINDOW_SIZE);
333             if (newInitialWindowSize >= 0) {
334                 updateInitialSendWindowSize(newInitialWindowSize);
335             }
336 
337         } else if (msg instanceof SpdyPingFrame) {
338 
339             /*
340              * SPDY PING frame processing requirements:
341              *
342              * Receivers of a PING frame should send an identical frame to the sender
343              * as soon as possible.
344              *
345              * Receivers of a PING frame must ignore frames that it did not initiate
346              */
347 
348             SpdyPingFrame spdyPingFrame = (SpdyPingFrame) msg;
349 
350             if (isRemoteInitiatedId(spdyPingFrame.getId())) {
351                 Channels.write(ctx, Channels.future(e.getChannel()), spdyPingFrame, e.getRemoteAddress());
352                 return;
353             }
354 
355             // Note: only checks that there are outstanding pings since uniqueness is not enforced
356             if (pings.get() == 0) {
357                 return;
358             }
359             pings.getAndDecrement();
360 
361         } else if (msg instanceof SpdyGoAwayFrame) {
362 
363             receivedGoAwayFrame = true;
364 
365         } else if (msg instanceof SpdyHeadersFrame) {
366 
367             SpdyHeadersFrame spdyHeadersFrame = (SpdyHeadersFrame) msg;
368             int streamId = spdyHeadersFrame.getStreamId();
369 
370             // Check if we received a valid HEADERS frame
371             if (spdyHeadersFrame.isInvalid()) {
372                 issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.PROTOCOL_ERROR);
373                 return;
374             }
375 
376             if (spdySession.isRemoteSideClosed(streamId)) {
377                 issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.INVALID_STREAM);
378                 return;
379             }
380 
381             // Close the remote side of the stream if this is the last frame
382             if (spdyHeadersFrame.isLast()) {
383                 halfCloseStream(streamId, true, e.getFuture());
384             }
385 
386         } else if (msg instanceof SpdyWindowUpdateFrame) {
387 
388             /*
389              * SPDY WINDOW_UPDATE frame processing requirements:
390              *
391              * Receivers of a WINDOW_UPDATE that cause the window size to exceed 2^31
392              * must send a RST_STREAM with the status code FLOW_CONTROL_ERROR.
393              *
394              * Sender should ignore all WINDOW_UPDATE frames associated with a stream
395              * after sending the last frame for the stream.
396              */
397 
398             SpdyWindowUpdateFrame spdyWindowUpdateFrame = (SpdyWindowUpdateFrame) msg;
399             int streamId = spdyWindowUpdateFrame.getStreamId();
400             int deltaWindowSize = spdyWindowUpdateFrame.getDeltaWindowSize();
401 
402             // Ignore frames for half-closed streams
403             if (streamId != SPDY_SESSION_STREAM_ID && spdySession.isLocalSideClosed(streamId)) {
404                 return;
405             }
406 
407             // Check for numerical overflow
408             if (spdySession.getSendWindowSize(streamId) > Integer.MAX_VALUE - deltaWindowSize) {
409                 if (streamId == SPDY_SESSION_STREAM_ID) {
410                     issueSessionError(ctx, e.getChannel(), e.getRemoteAddress(), SpdySessionStatus.PROTOCOL_ERROR);
411                 } else {
412                     issueStreamError(ctx, e.getRemoteAddress(), streamId, SpdyStreamStatus.FLOW_CONTROL_ERROR);
413                 }
414                 return;
415             }
416 
417             updateSendWindowSize(ctx, streamId, deltaWindowSize);
418             return;
419         }
420 
421         super.messageReceived(ctx, e);
422     }
423 
424     @Override
425     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
426             throws Exception {
427 
428         Throwable cause = e.getCause();
429         if (cause instanceof SpdyProtocolException) {
430             issueSessionError(ctx, e.getChannel(), null, SpdySessionStatus.PROTOCOL_ERROR);
431         }
432 
433         super.exceptionCaught(ctx, e);
434     }
435 
436     public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent evt)
437             throws Exception {
438         if (evt instanceof ChannelStateEvent) {
439             ChannelStateEvent e = (ChannelStateEvent) evt;
440             switch (e.getState()) {
441             case OPEN:
442             case CONNECTED:
443             case BOUND:
444 
445                 /*
446                  * SPDY connection requirements:
447                  *
448                  * When either endpoint closes the transport-level connection,
449                  * it must first send a GOAWAY frame.
450                  */
451                 if (Boolean.FALSE.equals(e.getValue()) || e.getValue() == null) {
452                     sendGoAwayFrame(ctx, e);
453                     return;
454                 }
455             }
456         }
457         if (!(evt instanceof MessageEvent)) {
458             ctx.sendDownstream(evt);
459             return;
460         }
461 
462         MessageEvent e = (MessageEvent) evt;
463         Object msg = e.getMessage();
464 
465         if (msg instanceof SpdyDataFrame) {
466 
467             SpdyDataFrame spdyDataFrame = (SpdyDataFrame) msg;
468             final int streamId = spdyDataFrame.getStreamId();
469 
470             // Frames must not be sent on half-closed streams
471             if (spdySession.isLocalSideClosed(streamId)) {
472                 e.getFuture().setFailure(PROTOCOL_EXCEPTION);
473                 return;
474             }
475 
476             /*
477              * SPDY Data frame flow control processing requirements:
478              *
479              * Sender must not send a data frame with data length greater
480              * than the transfer window size.
481              *
482              * After sending each data frame, the sender decrements its
483              * transfer window size by the amount of data transmitted.
484              *
485              * When the window size becomes less than or equal to 0, the
486              * sender must pause transmitting data frames.
487              */
488 
489             synchronized (flowControlLock) {
490                 int dataLength = spdyDataFrame.getData().readableBytes();
491                 int sendWindowSize = spdySession.getSendWindowSize(streamId);
492                 int sessionSendWindowSize = spdySession.getSendWindowSize(SPDY_SESSION_STREAM_ID);
493                 sendWindowSize = Math.min(sendWindowSize, sessionSendWindowSize);
494 
495                 if (sendWindowSize <= 0) {
496                     // Stream is stalled -- enqueue Data frame and return
497                     spdySession.putPendingWrite(streamId, e);
498                     return;
499                 } else if (sendWindowSize < dataLength) {
500                     // Stream is not stalled but we cannot send the entire frame
501                     spdySession.updateSendWindowSize(streamId, -1 * sendWindowSize);
502                     spdySession.updateSendWindowSize(SPDY_SESSION_STREAM_ID, -1 * sendWindowSize);
503 
504                     // Create a partial data frame whose length is the current window size
505                     SpdyDataFrame partialDataFrame = new DefaultSpdyDataFrame(streamId);
506                     partialDataFrame.setData(spdyDataFrame.getData().readSlice(sendWindowSize));
507 
508                     // Enqueue the remaining data (will be the first frame queued)
509                     spdySession.putPendingWrite(streamId, e);
510 
511                     ChannelFuture writeFuture = Channels.future(e.getChannel());
512 
513                     // The transfer window size is pre-decremented when sending a data frame downstream.
514                     // Close the session on write failures that leaves the transfer window in a corrupt state.
515                     final SocketAddress remoteAddress = e.getRemoteAddress();
516                     final ChannelHandlerContext context = ctx;
517                     e.getFuture().addListener(new ChannelFutureListener() {
518                         public void operationComplete(ChannelFuture future) throws Exception {
519                             if (!future.isSuccess()) {
520                                 Channel channel = future.getChannel();
521                                 issueSessionError(
522                                         context, channel, remoteAddress, SpdySessionStatus.INTERNAL_ERROR);
523                             }
524                         }
525                     });
526 
527                     Channels.write(ctx, writeFuture, partialDataFrame, remoteAddress);
528                     return;
529                 } else {
530                     // Window size is large enough to send entire data frame
531                     spdySession.updateSendWindowSize(streamId, -1 * dataLength);
532                     spdySession.updateSendWindowSize(SPDY_SESSION_STREAM_ID, -1 * dataLength);
533 
534                     // The transfer window size is pre-decremented when sending a data frame downstream.
535                     // Close the session on write failures that leaves the transfer window in a corrupt state.
536                     final SocketAddress remoteAddress = e.getRemoteAddress();
537                     final ChannelHandlerContext context = ctx;
538                     e.getFuture().addListener(new ChannelFutureListener() {
539                         public void operationComplete(ChannelFuture future) throws Exception {
540                             if (!future.isSuccess()) {
541                                 Channel channel = future.getChannel();
542                                 issueSessionError(
543                                         context, channel, remoteAddress, SpdySessionStatus.INTERNAL_ERROR);
544                             }
545                         }
546                     });
547                 }
548             }
549 
550             // Close the local side of the stream if this is the last frame
551             if (spdyDataFrame.isLast()) {
552                 halfCloseStream(streamId, false, e.getFuture());
553             }
554 
555         } else if (msg instanceof SpdySynStreamFrame) {
556 
557             SpdySynStreamFrame spdySynStreamFrame = (SpdySynStreamFrame) msg;
558             int streamId = spdySynStreamFrame.getStreamId();
559 
560             if (isRemoteInitiatedId(streamId)) {
561                 e.getFuture().setFailure(PROTOCOL_EXCEPTION);
562                 return;
563             }
564 
565             byte priority = spdySynStreamFrame.getPriority();
566             boolean remoteSideClosed = spdySynStreamFrame.isUnidirectional();
567             boolean localSideClosed = spdySynStreamFrame.isLast();
568             if (!acceptStream(streamId, priority, remoteSideClosed, localSideClosed)) {
569                 e.getFuture().setFailure(PROTOCOL_EXCEPTION);
570                 return;
571             }
572 
573         } else if (msg instanceof SpdySynReplyFrame) {
574 
575             SpdySynReplyFrame spdySynReplyFrame = (SpdySynReplyFrame) msg;
576             int streamId = spdySynReplyFrame.getStreamId();
577 
578             // Frames must not be sent on half-closed streams
579             if (!isRemoteInitiatedId(streamId) || spdySession.isLocalSideClosed(streamId)) {
580                 e.getFuture().setFailure(PROTOCOL_EXCEPTION);
581                 return;
582             }
583 
584             // Close the local side of the stream if this is the last frame
585             if (spdySynReplyFrame.isLast()) {
586                 halfCloseStream(streamId, false, e.getFuture());
587             }
588 
589         } else if (msg instanceof SpdyRstStreamFrame) {
590 
591             SpdyRstStreamFrame spdyRstStreamFrame = (SpdyRstStreamFrame) msg;
592             removeStream(spdyRstStreamFrame.getStreamId(), e.getFuture());
593 
594         } else if (msg instanceof SpdySettingsFrame) {
595 
596             SpdySettingsFrame spdySettingsFrame = (SpdySettingsFrame) msg;
597 
598             int settingsMinorVersion = spdySettingsFrame.getValue(SpdySettingsFrame.SETTINGS_MINOR_VERSION);
599             if (settingsMinorVersion >= 0 && settingsMinorVersion != minorVersion) {
600                 // Settings frame had the wrong minor version
601                 e.getFuture().setFailure(PROTOCOL_EXCEPTION);
602                 return;
603             }
604 
605             int newConcurrentStreams =
606                     spdySettingsFrame.getValue(SpdySettingsFrame.SETTINGS_MAX_CONCURRENT_STREAMS);
607             if (newConcurrentStreams >= 0) {
608                 localConcurrentStreams = newConcurrentStreams;
609             }
610 
611             // Persistence flag are inconsistent with the use of SETTINGS to communicate
612             // the initial window size. Remove flags from the sender requesting that the
613             // value be persisted. Remove values that the sender indicates are persisted.
614             if (spdySettingsFrame.isPersisted(SpdySettingsFrame.SETTINGS_INITIAL_WINDOW_SIZE)) {
615                 spdySettingsFrame.removeValue(SpdySettingsFrame.SETTINGS_INITIAL_WINDOW_SIZE);
616             }
617             spdySettingsFrame.setPersistValue(SpdySettingsFrame.SETTINGS_INITIAL_WINDOW_SIZE, false);
618 
619             int newInitialWindowSize =
620                     spdySettingsFrame.getValue(SpdySettingsFrame.SETTINGS_INITIAL_WINDOW_SIZE);
621             if (newInitialWindowSize >= 0) {
622                 updateInitialReceiveWindowSize(newInitialWindowSize);
623             }
624 
625         } else if (msg instanceof SpdyPingFrame) {
626 
627             SpdyPingFrame spdyPingFrame = (SpdyPingFrame) msg;
628             if (isRemoteInitiatedId(spdyPingFrame.getId())) {
629                 e.getFuture().setFailure(new IllegalArgumentException(
630                             "invalid PING ID: " + spdyPingFrame.getId()));
631                 return;
632             }
633             pings.getAndIncrement();
634 
635         } else if (msg instanceof SpdyGoAwayFrame) {
636 
637             // Why is this being sent? Intercept it and fail the write.
638             // Should have sent a CLOSE ChannelStateEvent
639             e.getFuture().setFailure(PROTOCOL_EXCEPTION);
640             return;
641 
642         } else if (msg instanceof SpdyHeadersFrame) {
643 
644             SpdyHeadersFrame spdyHeadersFrame = (SpdyHeadersFrame) msg;
645             int streamId = spdyHeadersFrame.getStreamId();
646 
647             // Frames must not be sent on half-closed streams
648             if (spdySession.isLocalSideClosed(streamId)) {
649                 e.getFuture().setFailure(PROTOCOL_EXCEPTION);
650                 return;
651             }
652 
653             // Close the local side of the stream if this is the last frame
654             if (spdyHeadersFrame.isLast()) {
655                 halfCloseStream(streamId, false, e.getFuture());
656             }
657 
658         } else if (msg instanceof SpdyWindowUpdateFrame) {
659 
660             // Why is this being sent? Intercept it and fail the write.
661             e.getFuture().setFailure(PROTOCOL_EXCEPTION);
662             return;
663         }
664 
665         ctx.sendDownstream(evt);
666     }
667 
668     /*
669      * SPDY Session Error Handling:
670      *
671      * When a session error occurs, the endpoint encountering the error must first
672      * send a GOAWAY frame with the Stream-ID of the most recently received stream
673      * from the remote endpoint, and the error code for why the session is terminating.
674      *
675      * After sending the GOAWAY frame, the endpoint must close the TCP connection.
676      */
677     private void issueSessionError(
678             ChannelHandlerContext ctx, Channel channel, SocketAddress remoteAddress, SpdySessionStatus status) {
679 
680         ChannelFuture future = sendGoAwayFrame(ctx, channel, remoteAddress, status);
681         future.addListener(ChannelFutureListener.CLOSE);
682     }
683 
684     /*
685      * SPDY Stream Error Handling:
686      *
687      * Upon a stream error, the endpoint must send a RST_STREAM frame which contains
688      * the Stream-ID for the stream where the error occurred and the error status which
689      * caused the error.
690      *
691      * After sending the RST_STREAM, the stream is closed to the sending endpoint.
692      *
693      * Note: this is only called by the worker thread
694      */
695     private void issueStreamError(
696             ChannelHandlerContext ctx, SocketAddress remoteAddress, int streamId, SpdyStreamStatus status) {
697 
698         boolean fireMessageReceived = !spdySession.isRemoteSideClosed(streamId);
699         ChannelFuture future = Channels.future(ctx.getChannel());
700         removeStream(streamId, future);
701 
702         SpdyRstStreamFrame spdyRstStreamFrame = new DefaultSpdyRstStreamFrame(streamId, status);
703         Channels.write(ctx, future, spdyRstStreamFrame, remoteAddress);
704         if (fireMessageReceived) {
705             Channels.fireMessageReceived(ctx, spdyRstStreamFrame, remoteAddress);
706         }
707     }
708 
709     /*
710      * Helper functions
711      */
712 
713     private boolean isRemoteInitiatedId(int id) {
714         boolean serverId = isServerId(id);
715         return server && !serverId || !server && serverId;
716     }
717 
718     // need to synchronize to prevent new streams from being created while updating active streams
719     private synchronized void updateInitialSendWindowSize(int newInitialWindowSize) {
720         int deltaWindowSize = newInitialWindowSize - initialSendWindowSize;
721         initialSendWindowSize = newInitialWindowSize;
722         spdySession.updateAllSendWindowSizes(deltaWindowSize);
723     }
724 
725     // need to synchronize to prevent new streams from being created while updating active streams
726     private synchronized void updateInitialReceiveWindowSize(int newInitialWindowSize) {
727         int deltaWindowSize = newInitialWindowSize - initialReceiveWindowSize;
728         initialReceiveWindowSize = newInitialWindowSize;
729         spdySession.updateAllReceiveWindowSizes(deltaWindowSize);
730     }
731 
732     // need to synchronize accesses to sentGoAwayFrame, lastGoodStreamId, and initial window sizes
733     private synchronized boolean acceptStream(
734             int streamId, byte priority, boolean remoteSideClosed, boolean localSideClosed) {
735         // Cannot initiate any new streams after receiving or sending GOAWAY
736         if (receivedGoAwayFrame || sentGoAwayFrame) {
737             return false;
738         }
739 
740         boolean remote = isRemoteInitiatedId(streamId);
741         int maxConcurrentStreams = remote ? localConcurrentStreams : remoteConcurrentStreams;
742         if (spdySession.numActiveStreams(remote) >= maxConcurrentStreams) {
743             return false;
744         }
745         spdySession.acceptStream(
746                 streamId, priority, remoteSideClosed, localSideClosed,
747                 initialSendWindowSize, initialReceiveWindowSize, remote);
748         if (remote) {
749             lastGoodStreamId = streamId;
750         }
751         return true;
752     }
753 
754     private void halfCloseStream(int streamId, boolean remote, ChannelFuture future) {
755         if (remote) {
756             spdySession.closeRemoteSide(streamId, isRemoteInitiatedId(streamId));
757         } else {
758             spdySession.closeLocalSide(streamId, isRemoteInitiatedId(streamId));
759         }
760         if (closeSessionFutureListener != null && spdySession.noActiveStreams()) {
761             future.addListener(closeSessionFutureListener);
762         }
763     }
764 
765     private void removeStream(int streamId, ChannelFuture future) {
766         spdySession.removeStream(streamId, isRemoteInitiatedId(streamId));
767         if (closeSessionFutureListener != null && spdySession.noActiveStreams()) {
768             future.addListener(closeSessionFutureListener);
769         }
770     }
771 
772     private void updateSendWindowSize(ChannelHandlerContext ctx, int streamId, int deltaWindowSize) {
773         synchronized (flowControlLock) {
774             int newWindowSize = spdySession.updateSendWindowSize(streamId, deltaWindowSize);
775             if (streamId != SPDY_SESSION_STREAM_ID) {
776                 int sessionSendWindowSize = spdySession.getSendWindowSize(SPDY_SESSION_STREAM_ID);
777                 newWindowSize = Math.min(newWindowSize, sessionSendWindowSize);
778             }
779 
780             while (newWindowSize > 0) {
781                 // Check if we have unblocked a stalled stream
782                 MessageEvent e = spdySession.getPendingWrite(streamId);
783                 if (e == null) {
784                     break;
785                 }
786 
787                 SpdyDataFrame spdyDataFrame = (SpdyDataFrame) e.getMessage();
788                 int dataFrameSize = spdyDataFrame.getData().readableBytes();
789                 int writeStreamId = spdyDataFrame.getStreamId();
790                 if (streamId == SPDY_SESSION_STREAM_ID) {
791                     newWindowSize = Math.min(newWindowSize, spdySession.getSendWindowSize(writeStreamId));
792                 }
793 
794                 if (newWindowSize >= dataFrameSize) {
795                     // Window size is large enough to send entire data frame
796                     spdySession.removePendingWrite(writeStreamId);
797                     newWindowSize = spdySession.updateSendWindowSize(writeStreamId, -1 * dataFrameSize);
798                     int sessionSendWindowSize =
799                             spdySession.updateSendWindowSize(SPDY_SESSION_STREAM_ID, -1 * dataFrameSize);
800                     newWindowSize = Math.min(newWindowSize, sessionSendWindowSize);
801 
802                     // The transfer window size is pre-decremented when sending a data frame downstream.
803                     // Close the session on write failures that leaves the transfer window in a corrupt state.
804                     final SocketAddress remoteAddress = e.getRemoteAddress();
805                     final ChannelHandlerContext context = ctx;
806                     e.getFuture().addListener(new ChannelFutureListener() {
807                         public void operationComplete(ChannelFuture future) throws Exception {
808                             if (!future.isSuccess()) {
809                                 Channel channel = future.getChannel();
810                                 issueSessionError(context, channel, remoteAddress, SpdySessionStatus.INTERNAL_ERROR);
811                             }
812                         }
813                     });
814 
815                     // Close the local side of the stream if this is the last frame
816                     if (spdyDataFrame.isLast()) {
817                         halfCloseStream(writeStreamId, false, e.getFuture());
818                     }
819 
820                     Channels.write(ctx, e.getFuture(), spdyDataFrame, e.getRemoteAddress());
821                 } else {
822                     // We can send a partial frame
823                     spdySession.updateSendWindowSize(writeStreamId, -1 * newWindowSize);
824                     spdySession.updateSendWindowSize(SPDY_SESSION_STREAM_ID, -1 * newWindowSize);
825 
826                     // Create a partial data frame whose length is the current window size
827                     SpdyDataFrame partialDataFrame = new DefaultSpdyDataFrame(writeStreamId);
828                     partialDataFrame.setData(spdyDataFrame.getData().readSlice(newWindowSize));
829 
830                     ChannelFuture writeFuture = Channels.future(e.getChannel());
831 
832                     // The transfer window size is pre-decremented when sending a data frame downstream.
833                     // Close the session on write failures that leaves the transfer window in a corrupt state.
834                     final SocketAddress remoteAddress = e.getRemoteAddress();
835                     final ChannelHandlerContext context = ctx;
836                     e.getFuture().addListener(new ChannelFutureListener() {
837                         public void operationComplete(ChannelFuture future) throws Exception {
838                             if (!future.isSuccess()) {
839                                 Channel channel = future.getChannel();
840                                 issueSessionError(context, channel, remoteAddress, SpdySessionStatus.INTERNAL_ERROR);
841                             }
842                         }
843                     });
844 
845                     Channels.write(ctx, writeFuture, partialDataFrame, remoteAddress);
846 
847                     newWindowSize = 0;
848                 }
849             }
850         }
851     }
852 
853     private void sendGoAwayFrame(ChannelHandlerContext ctx, ChannelStateEvent e) {
854         // Avoid NotYetConnectedException
855         if (!e.getChannel().isConnected()) {
856             ctx.sendDownstream(e);
857             return;
858         }
859 
860         ChannelFuture future = sendGoAwayFrame(ctx, e.getChannel(), null, SpdySessionStatus.OK);
861         if (spdySession.noActiveStreams()) {
862             future.addListener(new ClosingChannelFutureListener(ctx, e));
863         } else {
864             closeSessionFutureListener = new ClosingChannelFutureListener(ctx, e);
865         }
866     }
867 
868     private synchronized ChannelFuture sendGoAwayFrame(
869             ChannelHandlerContext ctx, Channel channel, SocketAddress remoteAddress, SpdySessionStatus status) {
870         if (!sentGoAwayFrame) {
871             sentGoAwayFrame = true;
872             SpdyGoAwayFrame spdyGoAwayFrame = new DefaultSpdyGoAwayFrame(lastGoodStreamId, status);
873             ChannelFuture future = Channels.future(channel);
874             Channels.write(ctx, future, spdyGoAwayFrame, remoteAddress);
875             return future;
876         }
877         return Channels.succeededFuture(channel);
878     }
879 
880     private static final class ClosingChannelFutureListener implements ChannelFutureListener {
881         private final ChannelHandlerContext ctx;
882         private final ChannelStateEvent e;
883 
884         ClosingChannelFutureListener(ChannelHandlerContext ctx, ChannelStateEvent e) {
885             this.ctx = ctx;
886             this.e = e;
887         }
888 
889         public void operationComplete(ChannelFuture sentGoAwayFuture) throws Exception {
890             if (!(sentGoAwayFuture.getCause() instanceof ClosedChannelException)) {
891                 Channels.close(ctx, e.getFuture());
892             } else {
893                 e.getFuture().setSuccess();
894             }
895         }
896     }
897 }