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