1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.spdy;
20
21 import java.io.IOException;
22 import java.net.InetSocketAddress;
23 import java.nio.ByteBuffer;
24 import java.nio.channels.InterruptedByTimeoutException;
25 import java.util.ArrayList;
26 import java.util.Collections;
27 import java.util.HashSet;
28 import java.util.LinkedList;
29 import java.util.List;
30 import java.util.Map;
31 import java.util.Objects;
32 import java.util.Set;
33 import java.util.concurrent.ConcurrentHashMap;
34 import java.util.concurrent.ConcurrentMap;
35 import java.util.concurrent.CopyOnWriteArrayList;
36 import java.util.concurrent.ExecutionException;
37 import java.util.concurrent.Executor;
38 import java.util.concurrent.TimeUnit;
39 import java.util.concurrent.TimeoutException;
40 import java.util.concurrent.atomic.AtomicBoolean;
41 import java.util.concurrent.atomic.AtomicInteger;
42
43 import org.eclipse.jetty.io.ByteBufferPool;
44 import org.eclipse.jetty.io.EndPoint;
45 import org.eclipse.jetty.spdy.api.ByteBufferDataInfo;
46 import org.eclipse.jetty.spdy.api.DataInfo;
47 import org.eclipse.jetty.spdy.api.GoAwayInfo;
48 import org.eclipse.jetty.spdy.api.GoAwayReceivedInfo;
49 import org.eclipse.jetty.spdy.api.PingInfo;
50 import org.eclipse.jetty.spdy.api.PingResultInfo;
51 import org.eclipse.jetty.spdy.api.RstInfo;
52 import org.eclipse.jetty.spdy.api.SPDYException;
53 import org.eclipse.jetty.spdy.api.Session;
54 import org.eclipse.jetty.spdy.api.SessionFrameListener;
55 import org.eclipse.jetty.spdy.api.SessionStatus;
56 import org.eclipse.jetty.spdy.api.Settings;
57 import org.eclipse.jetty.spdy.api.SettingsInfo;
58 import org.eclipse.jetty.spdy.api.Stream;
59 import org.eclipse.jetty.spdy.api.StreamFrameListener;
60 import org.eclipse.jetty.spdy.api.StreamStatus;
61 import org.eclipse.jetty.spdy.api.SynInfo;
62 import org.eclipse.jetty.spdy.frames.ControlFrame;
63 import org.eclipse.jetty.spdy.frames.ControlFrameType;
64 import org.eclipse.jetty.spdy.frames.CredentialFrame;
65 import org.eclipse.jetty.spdy.frames.DataFrame;
66 import org.eclipse.jetty.spdy.frames.GoAwayFrame;
67 import org.eclipse.jetty.spdy.frames.HeadersFrame;
68 import org.eclipse.jetty.spdy.frames.PingFrame;
69 import org.eclipse.jetty.spdy.frames.RstStreamFrame;
70 import org.eclipse.jetty.spdy.frames.SettingsFrame;
71 import org.eclipse.jetty.spdy.frames.SynReplyFrame;
72 import org.eclipse.jetty.spdy.frames.SynStreamFrame;
73 import org.eclipse.jetty.spdy.frames.WindowUpdateFrame;
74 import org.eclipse.jetty.spdy.generator.Generator;
75 import org.eclipse.jetty.spdy.parser.Parser;
76 import org.eclipse.jetty.util.Atomics;
77 import org.eclipse.jetty.util.BufferUtil;
78 import org.eclipse.jetty.util.Callback;
79 import org.eclipse.jetty.util.ForkInvoker;
80 import org.eclipse.jetty.util.FutureCallback;
81 import org.eclipse.jetty.util.FuturePromise;
82 import org.eclipse.jetty.util.Promise;
83 import org.eclipse.jetty.util.component.ContainerLifeCycle;
84 import org.eclipse.jetty.util.component.Dumpable;
85 import org.eclipse.jetty.util.log.Log;
86 import org.eclipse.jetty.util.log.Logger;
87 import org.eclipse.jetty.util.thread.Scheduler;
88
89 public class StandardSession implements ISession, Parser.Listener, Dumpable
90 {
91 private static final Logger LOG = Log.getLogger(Session.class);
92
93 private final ForkInvoker<Callback> invoker = new SessionInvoker();
94 private final Map<String, Object> attributes = new ConcurrentHashMap<>();
95 private final List<Listener> listeners = new CopyOnWriteArrayList<>();
96 private final ConcurrentMap<Integer, IStream> streams = new ConcurrentHashMap<>();
97 private final LinkedList<FrameBytes> queue = new LinkedList<>();
98 private final ByteBufferPool bufferPool;
99 private final Executor threadPool;
100 private final Scheduler scheduler;
101 private final short version;
102 private final Controller controller;
103 private final EndPoint endPoint;
104 private final IdleListener idleListener;
105 private final AtomicInteger streamIds;
106 private final AtomicInteger pingIds;
107 private final SessionFrameListener listener;
108 private final Generator generator;
109 private final AtomicBoolean goAwaySent = new AtomicBoolean();
110 private final AtomicBoolean goAwayReceived = new AtomicBoolean();
111 private final AtomicInteger lastStreamId = new AtomicInteger();
112 private final FlowControlStrategy flowControlStrategy;
113 private boolean flushing;
114 private Throwable failure;
115
116 public StandardSession(short version, ByteBufferPool bufferPool, Executor threadPool, Scheduler scheduler,
117 Controller controller, EndPoint endPoint, IdleListener idleListener, int initialStreamId,
118 SessionFrameListener listener, Generator generator, FlowControlStrategy flowControlStrategy)
119 {
120
121 this.version = version;
122 this.bufferPool = bufferPool;
123 this.threadPool = threadPool;
124 this.scheduler = scheduler;
125 this.controller = controller;
126 this.endPoint = endPoint;
127 this.idleListener = idleListener;
128 this.streamIds = new AtomicInteger(initialStreamId);
129 this.pingIds = new AtomicInteger(initialStreamId);
130 this.listener = listener;
131 this.generator = generator;
132 this.flowControlStrategy = flowControlStrategy;
133 }
134
135 @Override
136 public short getVersion()
137 {
138 return version;
139 }
140
141 @Override
142 public void addListener(Listener listener)
143 {
144 listeners.add(listener);
145 }
146
147 @Override
148 public void removeListener(Listener listener)
149 {
150 listeners.remove(listener);
151 }
152
153 @Override
154 public Stream syn(SynInfo synInfo, StreamFrameListener listener) throws ExecutionException, InterruptedException, TimeoutException
155 {
156 FuturePromise<Stream> result = new FuturePromise<>();
157 syn(synInfo, listener, result);
158 if (synInfo.getTimeout() > 0)
159 return result.get(synInfo.getTimeout(), synInfo.getUnit());
160 else
161 return result.get();
162 }
163
164 @Override
165 public void syn(SynInfo synInfo, StreamFrameListener listener, Promise<Stream> promise)
166 {
167
168
169
170
171
172
173 int associatedStreamId = 0;
174 if (synInfo instanceof PushSynInfo)
175 associatedStreamId = ((PushSynInfo)synInfo).getAssociatedStreamId();
176
177 synchronized (this)
178 {
179 int streamId = streamIds.getAndAdd(2);
180
181 SynStreamFrame synStream = new SynStreamFrame(version, synInfo.getFlags(), streamId, associatedStreamId, synInfo.getPriority(), (short)0, synInfo.getHeaders());
182 IStream stream = createStream(synStream, listener, true, promise);
183 generateAndEnqueueControlFrame(stream, synStream, synInfo.getTimeout(), synInfo.getUnit(), stream);
184 }
185 flush();
186 }
187
188 @Override
189 public void rst(RstInfo rstInfo) throws InterruptedException, ExecutionException, TimeoutException
190 {
191 FutureCallback result = new FutureCallback();
192 rst(rstInfo, result);
193 if (rstInfo.getTimeout() > 0)
194 result.get(rstInfo.getTimeout(), rstInfo.getUnit());
195 else
196 result.get();
197 }
198
199 @Override
200 public void rst(RstInfo rstInfo, Callback callback)
201 {
202
203 if (goAwaySent.get())
204 {
205 complete(callback);
206 }
207 else
208 {
209 int streamId = rstInfo.getStreamId();
210 IStream stream = streams.get(streamId);
211 RstStreamFrame frame = new RstStreamFrame(version, streamId, rstInfo.getStreamStatus().getCode(version));
212 control(stream, frame, rstInfo.getTimeout(), rstInfo.getUnit(), callback);
213 if (stream != null)
214 {
215 stream.process(frame);
216 removeStream(stream);
217 }
218 }
219 }
220
221 @Override
222 public void settings(SettingsInfo settingsInfo) throws ExecutionException, InterruptedException, TimeoutException
223 {
224 FutureCallback result = new FutureCallback();
225 settings(settingsInfo, result);
226 if (settingsInfo.getTimeout() > 0)
227 result.get(settingsInfo.getTimeout(), settingsInfo.getUnit());
228 else
229 result.get();
230 }
231
232 @Override
233 public void settings(SettingsInfo settingsInfo, Callback callback)
234 {
235 SettingsFrame frame = new SettingsFrame(version, settingsInfo.getFlags(), settingsInfo.getSettings());
236 control(null, frame, settingsInfo.getTimeout(), settingsInfo.getUnit(), callback);
237 }
238
239 @Override
240 public PingResultInfo ping(PingInfo pingInfo) throws ExecutionException, InterruptedException, TimeoutException
241 {
242
243 FuturePromise<PingResultInfo> result = new FuturePromise<>();
244 ping(pingInfo, result);
245 if (pingInfo.getTimeout() > 0)
246 return result.get(pingInfo.getTimeout(), pingInfo.getUnit());
247 else
248 return result.get();
249 }
250
251 @Override
252 public void ping(PingInfo pingInfo, Promise<PingResultInfo> promise)
253 {
254 int pingId = pingIds.getAndAdd(2);
255 PingInfoCallback pingInfoCallback = new PingInfoCallback(pingId, promise);
256 PingFrame frame = new PingFrame(version, pingId);
257 control(null, frame, pingInfo.getTimeout(), pingInfo.getUnit(), pingInfoCallback);
258 }
259
260 @Override
261 public void goAway(GoAwayInfo goAwayInfo) throws ExecutionException, InterruptedException, TimeoutException
262 {
263 goAway(goAwayInfo, SessionStatus.OK);
264 }
265
266 private void goAway(GoAwayInfo goAwayInfo, SessionStatus sessionStatus) throws ExecutionException, InterruptedException, TimeoutException
267 {
268 FutureCallback result = new FutureCallback();
269 goAway(sessionStatus, goAwayInfo.getTimeout(), goAwayInfo.getUnit(), result);
270 if (goAwayInfo.getTimeout() > 0)
271 result.get(goAwayInfo.getTimeout(), goAwayInfo.getUnit());
272 else
273 result.get();
274 }
275
276 @Override
277 public void goAway(GoAwayInfo goAwayInfo, Callback callback)
278 {
279 goAway(SessionStatus.OK, goAwayInfo.getTimeout(), goAwayInfo.getUnit(), callback);
280 }
281
282 private void goAway(SessionStatus sessionStatus, long timeout, TimeUnit unit, Callback callback)
283 {
284 if (goAwaySent.compareAndSet(false, true))
285 {
286 if (!goAwayReceived.get())
287 {
288 GoAwayFrame frame = new GoAwayFrame(version, lastStreamId.get(), sessionStatus.getCode());
289 control(null, frame, timeout, unit, callback);
290 return;
291 }
292 }
293 complete(callback);
294 }
295
296 @Override
297 public Set<Stream> getStreams()
298 {
299 Set<Stream> result = new HashSet<>();
300 result.addAll(streams.values());
301 return result;
302 }
303
304 @Override
305 public IStream getStream(int streamId)
306 {
307 return streams.get(streamId);
308 }
309
310 @Override
311 public Object getAttribute(String key)
312 {
313 return attributes.get(key);
314 }
315
316 @Override
317 public void setAttribute(String key, Object value)
318 {
319 attributes.put(key, value);
320 }
321
322 @Override
323 public Object removeAttribute(String key)
324 {
325 return attributes.remove(key);
326 }
327
328 @Override
329 public InetSocketAddress getLocalAddress()
330 {
331 return endPoint.getLocalAddress();
332 }
333
334 @Override
335 public InetSocketAddress getRemoteAddress()
336 {
337 return endPoint.getRemoteAddress();
338 }
339
340 @Override
341 public void onControlFrame(ControlFrame frame)
342 {
343 notifyIdle(idleListener, false);
344 try
345 {
346 LOG.debug("Processing {}", frame);
347
348 if (goAwaySent.get())
349 {
350 LOG.debug("Skipped processing of {}", frame);
351 return;
352 }
353
354 switch (frame.getType())
355 {
356 case SYN_STREAM:
357 {
358 onSyn((SynStreamFrame)frame);
359 break;
360 }
361 case SYN_REPLY:
362 {
363 onReply((SynReplyFrame)frame);
364 break;
365 }
366 case RST_STREAM:
367 {
368 onRst((RstStreamFrame)frame);
369 break;
370 }
371 case SETTINGS:
372 {
373 onSettings((SettingsFrame)frame);
374 break;
375 }
376 case NOOP:
377 {
378
379 break;
380 }
381 case PING:
382 {
383 onPing((PingFrame)frame);
384 break;
385 }
386 case GO_AWAY:
387 {
388 onGoAway((GoAwayFrame)frame);
389 break;
390 }
391 case HEADERS:
392 {
393 onHeaders((HeadersFrame)frame);
394 break;
395 }
396 case WINDOW_UPDATE:
397 {
398 onWindowUpdate((WindowUpdateFrame)frame);
399 break;
400 }
401 case CREDENTIAL:
402 {
403 onCredential((CredentialFrame)frame);
404 break;
405 }
406 default:
407 {
408 throw new IllegalStateException();
409 }
410 }
411 }
412 finally
413 {
414 notifyIdle(idleListener, true);
415 }
416 }
417
418 @Override
419 public void onDataFrame(DataFrame frame, ByteBuffer data)
420 {
421 notifyIdle(idleListener, false);
422 try
423 {
424 LOG.debug("Processing {}, {} data bytes", frame, data.remaining());
425
426 if (goAwaySent.get())
427 {
428 LOG.debug("Skipped processing of {}", frame);
429 return;
430 }
431
432 int streamId = frame.getStreamId();
433 IStream stream = streams.get(streamId);
434 if (stream == null)
435 {
436 RstInfo rstInfo = new RstInfo(streamId, StreamStatus.INVALID_STREAM);
437 LOG.debug("Unknown stream {}", rstInfo);
438 rst(rstInfo, new Callback.Adapter());
439 }
440 else
441 {
442 processData(stream, frame, data);
443 }
444 }
445 finally
446 {
447 notifyIdle(idleListener, true);
448 }
449 }
450
451 private void notifyIdle(IdleListener listener, boolean idle)
452 {
453 if (listener != null)
454 listener.onIdle(idle);
455 }
456
457 private void processData(final IStream stream, DataFrame frame, ByteBuffer data)
458 {
459 ByteBufferDataInfo dataInfo = new ByteBufferDataInfo(data, frame.isClose())
460 {
461 @Override
462 public void consume(int delta)
463 {
464 super.consume(delta);
465 flowControlStrategy.onDataConsumed(StandardSession.this, stream, this, delta);
466 }
467 };
468 flowControlStrategy.onDataReceived(this, stream, dataInfo);
469 stream.process(dataInfo);
470 if (stream.isClosed())
471 removeStream(stream);
472 }
473
474 @Override
475 public void onStreamException(StreamException x)
476 {
477 notifyOnException(listener, x);
478 rst(new RstInfo(x.getStreamId(), x.getStreamStatus()), new Callback.Adapter());
479 }
480
481 @Override
482 public void onSessionException(SessionException x)
483 {
484 Throwable cause = x.getCause();
485 notifyOnException(listener, cause == null ? x : cause);
486 goAway(x.getSessionStatus(), 0, TimeUnit.SECONDS, new Callback.Adapter());
487 }
488
489 private void onSyn(SynStreamFrame frame)
490 {
491 IStream stream = createStream(frame, null, false, null);
492 if (stream != null)
493 processSyn(listener, stream, frame);
494 }
495
496 private void processSyn(SessionFrameListener listener, IStream stream, SynStreamFrame frame)
497 {
498 stream.process(frame);
499
500 updateLastStreamId(stream);
501 SynInfo synInfo = new SynInfo(frame.getHeaders(), frame.isClose(), frame.getPriority());
502 StreamFrameListener streamListener = notifyOnSyn(listener, stream, synInfo);
503 stream.setStreamFrameListener(streamListener);
504 flush();
505
506 if (stream.isClosed())
507 removeStream(stream);
508 }
509
510 private IStream createStream(SynStreamFrame frame, StreamFrameListener listener, boolean local, Promise<Stream> promise)
511 {
512 IStream associatedStream = streams.get(frame.getAssociatedStreamId());
513 IStream stream = new StandardStream(frame.getStreamId(), frame.getPriority(), this, associatedStream, promise);
514 flowControlStrategy.onNewStream(this, stream);
515
516 stream.updateCloseState(frame.isClose(), local);
517 stream.setStreamFrameListener(listener);
518
519 if (stream.isUnidirectional())
520 {
521
522 stream.updateCloseState(true, !local);
523 if (!stream.isClosed())
524 stream.getAssociatedStream().associate(stream);
525 }
526
527 int streamId = stream.getId();
528 if (streams.putIfAbsent(streamId, stream) != null)
529 {
530 if (local)
531 throw new IllegalStateException("Duplicate stream id " + streamId);
532 RstInfo rstInfo = new RstInfo(streamId, StreamStatus.PROTOCOL_ERROR);
533 LOG.debug("Duplicate stream, {}", rstInfo);
534 try
535 {
536 rst(rstInfo);
537 }
538 catch (InterruptedException | ExecutionException | TimeoutException e)
539 {
540 e.printStackTrace();
541 }
542 return null;
543 }
544 else
545 {
546 LOG.debug("Created {}", stream);
547 if (local)
548 notifyStreamCreated(stream);
549 return stream;
550 }
551 }
552
553 private void notifyStreamCreated(IStream stream)
554 {
555 for (Listener listener : listeners)
556 {
557 if (listener instanceof StreamListener)
558 {
559 try
560 {
561 ((StreamListener)listener).onStreamCreated(stream);
562 }
563 catch (Exception x)
564 {
565 LOG.info("Exception while notifying listener " + listener, x);
566 }
567 catch (Error x)
568 {
569 LOG.info("Exception while notifying listener " + listener, x);
570 throw x;
571 }
572 }
573 }
574 }
575
576 private void removeStream(IStream stream)
577 {
578 if (stream.isUnidirectional())
579 stream.getAssociatedStream().disassociate(stream);
580
581 IStream removed = streams.remove(stream.getId());
582 if (removed != null)
583 assert removed == stream;
584
585 LOG.debug("Removed {}", stream);
586 notifyStreamClosed(stream);
587 }
588
589 private void notifyStreamClosed(IStream stream)
590 {
591 for (Listener listener : listeners)
592 {
593 if (listener instanceof StreamListener)
594 {
595 try
596 {
597 ((StreamListener)listener).onStreamClosed(stream);
598 }
599 catch (Exception x)
600 {
601 LOG.info("Exception while notifying listener " + listener, x);
602 }
603 catch (Error x)
604 {
605 LOG.info("Exception while notifying listener " + listener, x);
606 throw x;
607 }
608 }
609 }
610 }
611
612 private void onReply(SynReplyFrame frame)
613 {
614 int streamId = frame.getStreamId();
615 IStream stream = streams.get(streamId);
616 if (stream == null)
617 {
618 RstInfo rstInfo = new RstInfo(streamId, StreamStatus.INVALID_STREAM);
619 LOG.debug("Unknown stream {}", rstInfo);
620 rst(rstInfo, new Callback.Adapter());
621 }
622 else
623 {
624 processReply(stream, frame);
625 }
626 }
627
628 private void processReply(IStream stream, SynReplyFrame frame)
629 {
630 stream.process(frame);
631 if (stream.isClosed())
632 removeStream(stream);
633 }
634
635 private void onRst(RstStreamFrame frame)
636 {
637 IStream stream = streams.get(frame.getStreamId());
638
639 if (stream != null)
640 stream.process(frame);
641
642 RstInfo rstInfo = new RstInfo(frame.getStreamId(), StreamStatus.from(frame.getVersion(), frame.getStatusCode()));
643 notifyOnRst(listener, rstInfo);
644 flush();
645
646 if (stream != null)
647 removeStream(stream);
648 }
649
650 private void onSettings(SettingsFrame frame)
651 {
652 Settings.Setting windowSizeSetting = frame.getSettings().get(Settings.ID.INITIAL_WINDOW_SIZE);
653 if (windowSizeSetting != null)
654 {
655 int windowSize = windowSizeSetting.value();
656 setWindowSize(windowSize);
657 LOG.debug("Updated session window size to {}", windowSize);
658 }
659 SettingsInfo settingsInfo = new SettingsInfo(frame.getSettings(), frame.isClearPersisted());
660 notifyOnSettings(listener, settingsInfo);
661 flush();
662 }
663
664 private void onPing(PingFrame frame)
665 {
666 int pingId = frame.getPingId();
667 if (pingId % 2 == pingIds.get() % 2)
668 {
669 PingResultInfo pingResultInfo = new PingResultInfo(frame.getPingId());
670 notifyOnPing(listener, pingResultInfo);
671 flush();
672 }
673 else
674 {
675 control(null, frame, 0, TimeUnit.MILLISECONDS, new Callback.Adapter());
676 }
677 }
678
679 private void onGoAway(GoAwayFrame frame)
680 {
681 if (goAwayReceived.compareAndSet(false, true))
682 {
683
684 GoAwayReceivedInfo goAwayReceivedInfo = new GoAwayReceivedInfo(frame.getLastStreamId(), SessionStatus.from(frame.getStatusCode()));
685 notifyOnGoAway(listener, goAwayReceivedInfo);
686 flush();
687
688
689
690 }
691 }
692
693 private void onHeaders(HeadersFrame frame)
694 {
695 int streamId = frame.getStreamId();
696 IStream stream = streams.get(streamId);
697 if (stream == null)
698 {
699 RstInfo rstInfo = new RstInfo(streamId, StreamStatus.INVALID_STREAM);
700 LOG.debug("Unknown stream, {}", rstInfo);
701 rst(rstInfo, new Callback.Adapter());
702 }
703 else
704 {
705 processHeaders(stream, frame);
706 }
707 }
708
709 private void processHeaders(IStream stream, HeadersFrame frame)
710 {
711 stream.process(frame);
712 if (stream.isClosed())
713 removeStream(stream);
714 }
715
716 private void onWindowUpdate(WindowUpdateFrame frame)
717 {
718 int streamId = frame.getStreamId();
719 IStream stream = streams.get(streamId);
720 flowControlStrategy.onWindowUpdate(this, stream, frame.getWindowDelta());
721 flush();
722 }
723
724 private void onCredential(CredentialFrame frame)
725 {
726 LOG.warn("{} frame not yet supported", frame.getType());
727 flush();
728 }
729
730 protected void close()
731 {
732
733 if (controller != null)
734 controller.close(false);
735 }
736
737 private void notifyOnException(SessionFrameListener listener, Throwable x)
738 {
739 try
740 {
741 if (listener != null)
742 {
743 LOG.debug("Invoking callback with {} on listener {}", x, listener);
744 listener.onException(x);
745 }
746 }
747 catch (Exception xx)
748 {
749 LOG.info("Exception while notifying listener " + listener, xx);
750 }
751 catch (Error xx)
752 {
753 LOG.info("Exception while notifying listener " + listener, xx);
754 throw xx;
755 }
756 }
757
758 private StreamFrameListener notifyOnSyn(SessionFrameListener listener, Stream stream, SynInfo synInfo)
759 {
760 try
761 {
762 if (listener == null)
763 return null;
764 LOG.debug("Invoking callback with {} on listener {}", synInfo, listener);
765 return listener.onSyn(stream, synInfo);
766 }
767 catch (Exception x)
768 {
769 LOG.info("Exception while notifying listener " + listener, x);
770 return null;
771 }
772 catch (Error x)
773 {
774 LOG.info("Exception while notifying listener " + listener, x);
775 throw x;
776 }
777 }
778
779 private void notifyOnRst(SessionFrameListener listener, RstInfo rstInfo)
780 {
781 try
782 {
783 if (listener != null)
784 {
785 LOG.debug("Invoking callback with {} on listener {}", rstInfo, listener);
786 listener.onRst(this, rstInfo);
787 }
788 }
789 catch (Exception x)
790 {
791 LOG.info("Exception while notifying listener " + listener, x);
792 }
793 catch (Error x)
794 {
795 LOG.info("Exception while notifying listener " + listener, x);
796 throw x;
797 }
798 }
799
800 private void notifyOnSettings(SessionFrameListener listener, SettingsInfo settingsInfo)
801 {
802 try
803 {
804 if (listener != null)
805 {
806 LOG.debug("Invoking callback with {} on listener {}", settingsInfo, listener);
807 listener.onSettings(this, settingsInfo);
808 }
809 }
810 catch (Exception x)
811 {
812 LOG.info("Exception while notifying listener " + listener, x);
813 }
814 catch (Error x)
815 {
816 LOG.info("Exception while notifying listener " + listener, x);
817 throw x;
818 }
819 }
820
821 private void notifyOnPing(SessionFrameListener listener, PingResultInfo pingResultInfo)
822 {
823 try
824 {
825 if (listener != null)
826 {
827 LOG.debug("Invoking callback with {} on listener {}", pingResultInfo, listener);
828 listener.onPing(this, pingResultInfo);
829 }
830 }
831 catch (Exception x)
832 {
833 LOG.info("Exception while notifying listener " + listener, x);
834 }
835 catch (Error x)
836 {
837 LOG.info("Exception while notifying listener " + listener, x);
838 throw x;
839 }
840 }
841
842 private void notifyOnGoAway(SessionFrameListener listener, GoAwayReceivedInfo goAwayReceivedInfo)
843 {
844 try
845 {
846 if (listener != null)
847 {
848 LOG.debug("Invoking callback with {} on listener {}", goAwayReceivedInfo, listener);
849 listener.onGoAway(this, goAwayReceivedInfo);
850 }
851 }
852 catch (Exception x)
853 {
854 LOG.info("Exception while notifying listener " + listener, x);
855 }
856 catch (Error x)
857 {
858 LOG.info("Exception while notifying listener " + listener, x);
859 throw x;
860 }
861 }
862
863
864 @Override
865 public void control(IStream stream, ControlFrame frame, long timeout, TimeUnit unit, Callback callback)
866 {
867 generateAndEnqueueControlFrame(stream, frame, timeout, unit, callback);
868 flush();
869 }
870
871 private void generateAndEnqueueControlFrame(IStream stream, ControlFrame frame, long timeout, TimeUnit unit, Callback callback)
872 {
873 try
874 {
875
876
877
878 synchronized (this)
879 {
880 ByteBuffer buffer = generator.control(frame);
881 LOG.debug("Queuing {} on {}", frame, stream);
882 ControlFrameBytes frameBytes = new ControlFrameBytes(stream, callback, frame, buffer);
883 if (timeout > 0)
884 frameBytes.task = scheduler.schedule(frameBytes, timeout, unit);
885
886
887 if (ControlFrameType.PING == frame.getType())
888 prepend(frameBytes);
889 else
890 append(frameBytes);
891 }
892 }
893 catch (Exception x)
894 {
895 notifyCallbackFailed(callback, x);
896 }
897 }
898
899 private void updateLastStreamId(IStream stream)
900 {
901 int streamId = stream.getId();
902 if (streamId % 2 != streamIds.get() % 2)
903 Atomics.updateMax(lastStreamId, streamId);
904 }
905
906 @Override
907 public void data(IStream stream, DataInfo dataInfo, long timeout, TimeUnit unit, Callback callback)
908 {
909 LOG.debug("Queuing {} on {}", dataInfo, stream);
910 DataFrameBytes frameBytes = new DataFrameBytes(stream, callback, dataInfo);
911 if (timeout > 0)
912 frameBytes.task = scheduler.schedule(frameBytes, timeout, unit);
913 append(frameBytes);
914 flush();
915 }
916
917 @Override
918 public void shutdown()
919 {
920 FrameBytes frameBytes = new CloseFrameBytes();
921 append(frameBytes);
922 flush();
923 }
924
925 private void execute(Runnable task)
926 {
927 threadPool.execute(task);
928 }
929
930 @Override
931 public void flush()
932 {
933 FrameBytes frameBytes = null;
934 ByteBuffer buffer = null;
935 synchronized (queue)
936 {
937 if (flushing || queue.isEmpty())
938 return;
939
940 Set<IStream> stalledStreams = null;
941 for (int i = 0; i < queue.size(); ++i)
942 {
943 frameBytes = queue.get(i);
944
945 IStream stream = frameBytes.getStream();
946 if (stream != null && stalledStreams != null && stalledStreams.contains(stream))
947 continue;
948
949 buffer = frameBytes.getByteBuffer();
950 if (buffer != null)
951 {
952 queue.remove(i);
953 if (stream != null && stream.isReset() && !(frameBytes instanceof ControlFrameBytes))
954 {
955 frameBytes.fail(new StreamException(stream.getId(), StreamStatus.INVALID_STREAM,
956 "Stream: " + stream + " is reset!"));
957 return;
958 }
959 break;
960 }
961
962 if (stalledStreams == null)
963 stalledStreams = new HashSet<>();
964 if (stream != null)
965 stalledStreams.add(stream);
966
967 LOG.debug("Flush stalled for {}, {} frame(s) in queue", frameBytes, queue.size());
968 }
969
970 if (buffer == null)
971 return;
972
973 flushing = true;
974 LOG.debug("Flushing {}, {} frame(s) in queue", frameBytes, queue.size());
975 }
976 write(buffer, frameBytes);
977 }
978
979 private void append(FrameBytes frameBytes)
980 {
981 Throwable failure;
982 synchronized (queue)
983 {
984 failure = this.failure;
985 if (failure == null)
986 {
987
988
989 if (frameBytes instanceof ControlFrameBytes)
990 queue.addLast(frameBytes);
991 else
992 {
993 int index = queue.size();
994 while (index > 0)
995 {
996 FrameBytes element = queue.get(index - 1);
997 if (element.compareTo(frameBytes) >= 0)
998 break;
999 --index;
1000 }
1001 queue.add(index, frameBytes);
1002 }
1003 }
1004 }
1005
1006 if (failure != null)
1007 frameBytes.fail(new SPDYException(failure));
1008 }
1009
1010 private void prepend(FrameBytes frameBytes)
1011 {
1012 Throwable failure;
1013 synchronized (queue)
1014 {
1015 failure = this.failure;
1016 if (failure == null)
1017 {
1018 int index = 0;
1019 while (index < queue.size())
1020 {
1021 FrameBytes element = queue.get(index);
1022 if (element.compareTo(frameBytes) <= 0)
1023 break;
1024 ++index;
1025 }
1026 queue.add(index, frameBytes);
1027 }
1028 }
1029
1030 if (failure != null)
1031 frameBytes.fail(new SPDYException(failure));
1032 }
1033
1034 protected void write(ByteBuffer buffer, Callback callback)
1035 {
1036 if (controller != null)
1037 {
1038 LOG.debug("Writing {} frame bytes of {}", buffer.remaining(), buffer.limit());
1039 controller.write(buffer, callback);
1040 }
1041 }
1042
1043 private void complete(final Callback callback)
1044 {
1045
1046
1047
1048
1049 invoker.invoke(callback);
1050 }
1051
1052 private void notifyCallbackFailed(Callback callback, Throwable x)
1053 {
1054 try
1055 {
1056 if (callback != null)
1057 callback.failed(x);
1058 }
1059 catch (Exception xx)
1060 {
1061 LOG.info("Exception while notifying callback " + callback, xx);
1062 }
1063 catch (Error xx)
1064 {
1065 LOG.info("Exception while notifying callback " + callback, xx);
1066 throw xx;
1067 }
1068 }
1069
1070 public int getWindowSize()
1071 {
1072 return flowControlStrategy.getWindowSize(this);
1073 }
1074
1075 public void setWindowSize(int initialWindowSize)
1076 {
1077 flowControlStrategy.setWindowSize(this, initialWindowSize);
1078 }
1079
1080 @Override
1081 public String toString()
1082 {
1083 return String.format("%s@%x{v%d,queueSize=%d,windowSize=%d,streams=%d}", getClass().getSimpleName(),
1084 hashCode(), version, queue.size(), getWindowSize(), streams.size());
1085 }
1086
1087 @Override
1088 public String dump()
1089 {
1090 return ContainerLifeCycle.dump(this);
1091 }
1092
1093 @Override
1094 public void dump(Appendable out, String indent) throws IOException
1095 {
1096 ContainerLifeCycle.dumpObject(out, this);
1097 ContainerLifeCycle.dump(out, indent, Collections.singletonList(controller), streams.values());
1098 }
1099
1100 private class SessionInvoker extends ForkInvoker<Callback>
1101 {
1102 private SessionInvoker()
1103 {
1104 super(4);
1105 }
1106
1107 @Override
1108 public void fork(final Callback callback)
1109 {
1110 execute(new Runnable()
1111 {
1112 @Override
1113 public void run()
1114 {
1115 callback.succeeded();
1116 flush();
1117 }
1118 });
1119 }
1120
1121 @Override
1122 public void call(Callback callback)
1123 {
1124 callback.succeeded();
1125 flush();
1126 }
1127 }
1128
1129 public interface FrameBytes extends Comparable<FrameBytes>, Callback
1130 {
1131 public IStream getStream();
1132
1133 public abstract ByteBuffer getByteBuffer();
1134
1135 public abstract void complete();
1136
1137 public abstract void fail(Throwable throwable);
1138 }
1139
1140 private abstract class AbstractFrameBytes implements FrameBytes, Runnable
1141 {
1142 private final IStream stream;
1143 private final Callback callback;
1144 protected volatile Scheduler.Task task;
1145
1146 protected AbstractFrameBytes(IStream stream, Callback callback)
1147 {
1148 this.stream = stream;
1149 this.callback = Objects.requireNonNull(callback);
1150 }
1151
1152 @Override
1153 public IStream getStream()
1154 {
1155 return stream;
1156 }
1157
1158 @Override
1159 public int compareTo(FrameBytes that)
1160 {
1161
1162
1163 IStream thisStream = getStream();
1164 IStream thatStream = that.getStream();
1165 if (thisStream == null)
1166 return thatStream == null ? 0 : -1;
1167 if (thatStream == null)
1168 return 1;
1169
1170 return thatStream.getPriority() - thisStream.getPriority();
1171 }
1172
1173 @Override
1174 public void complete()
1175 {
1176 cancelTask();
1177 StandardSession.this.complete(callback);
1178 }
1179
1180 @Override
1181 public void fail(Throwable x)
1182 {
1183 cancelTask();
1184 notifyCallbackFailed(callback, x);
1185 StandardSession.this.flush();
1186 }
1187
1188 private void cancelTask()
1189 {
1190 Scheduler.Task task = this.task;
1191 if (task != null)
1192 task.cancel();
1193 }
1194
1195 @Override
1196 public void run()
1197 {
1198 close();
1199 fail(new InterruptedByTimeoutException());
1200 }
1201
1202 @Override
1203 public void succeeded()
1204 {
1205 synchronized (queue)
1206 {
1207 if (LOG.isDebugEnabled())
1208 LOG.debug("Completed write of {}, {} frame(s) in queue", this, queue.size());
1209 flushing = false;
1210 }
1211 complete();
1212 }
1213
1214 @Override
1215 public void failed(Throwable x)
1216 {
1217 List<FrameBytes> frameBytesToFail = new ArrayList<>();
1218 frameBytesToFail.add(this);
1219
1220 synchronized (queue)
1221 {
1222 failure = x;
1223 if (LOG.isDebugEnabled())
1224 {
1225 String logMessage = String.format("Failed write of %s, failing all %d frame(s) in queue", this, queue.size());
1226 LOG.debug(logMessage, x);
1227 }
1228 frameBytesToFail.addAll(queue);
1229 queue.clear();
1230 flushing = false;
1231 }
1232
1233 for (FrameBytes fb : frameBytesToFail)
1234 fb.fail(x);
1235 }
1236 }
1237
1238 private class ControlFrameBytes extends AbstractFrameBytes
1239 {
1240 private final ControlFrame frame;
1241 private final ByteBuffer buffer;
1242
1243 private ControlFrameBytes(IStream stream, Callback callback, ControlFrame frame, ByteBuffer buffer)
1244 {
1245 super(stream, callback);
1246 this.frame = frame;
1247 this.buffer = buffer;
1248 }
1249
1250 @Override
1251 public ByteBuffer getByteBuffer()
1252 {
1253 return buffer;
1254 }
1255
1256 @Override
1257 public void complete()
1258 {
1259 bufferPool.release(buffer);
1260
1261 super.complete();
1262
1263 if (frame.getType() == ControlFrameType.GO_AWAY)
1264 {
1265
1266
1267 close();
1268 }
1269 IStream stream = getStream();
1270 if (stream != null && stream.isClosed())
1271 removeStream(stream);
1272 }
1273
1274 @Override
1275 public String toString()
1276 {
1277 return frame.toString();
1278 }
1279 }
1280
1281 private class DataFrameBytes extends AbstractFrameBytes
1282 {
1283 private final DataInfo dataInfo;
1284 private int size;
1285 private volatile ByteBuffer buffer;
1286
1287 private DataFrameBytes(IStream stream, Callback handler, DataInfo dataInfo)
1288 {
1289 super(stream, handler);
1290 this.dataInfo = dataInfo;
1291 }
1292
1293 @Override
1294 public ByteBuffer getByteBuffer()
1295 {
1296 try
1297 {
1298 IStream stream = getStream();
1299 int windowSize = stream.getWindowSize();
1300 if (windowSize <= 0)
1301 return null;
1302
1303 size = dataInfo.available();
1304 if (size > windowSize)
1305 size = windowSize;
1306
1307 buffer = generator.data(stream.getId(), size, dataInfo);
1308 return buffer;
1309 }
1310 catch (Throwable x)
1311 {
1312 fail(x);
1313 return null;
1314 }
1315 }
1316
1317 @Override
1318 public void complete()
1319 {
1320 bufferPool.release(buffer);
1321 IStream stream = getStream();
1322 flowControlStrategy.updateWindow(StandardSession.this, stream, -size);
1323 if (dataInfo.available() > 0)
1324 {
1325
1326
1327
1328 prepend(this);
1329 flush();
1330 }
1331 else
1332 {
1333 super.complete();
1334 stream.updateCloseState(dataInfo.isClose(), true);
1335 if (stream.isClosed())
1336 removeStream(stream);
1337 }
1338 }
1339
1340 @Override
1341 public String toString()
1342 {
1343 return String.format("DATA bytes @%x available=%d consumed=%d on %s", dataInfo.hashCode(), dataInfo.available(), dataInfo.consumed(), getStream());
1344 }
1345 }
1346
1347 private class CloseFrameBytes extends AbstractFrameBytes
1348 {
1349 private CloseFrameBytes()
1350 {
1351 super(null, new Callback.Adapter());
1352 }
1353
1354 @Override
1355 public ByteBuffer getByteBuffer()
1356 {
1357 return BufferUtil.EMPTY_BUFFER;
1358 }
1359
1360 @Override
1361 public void complete()
1362 {
1363 super.complete();
1364 close();
1365 }
1366 }
1367
1368 private static class PingInfoCallback extends PingResultInfo implements Callback
1369 {
1370 private final Promise<PingResultInfo> promise;
1371
1372 public PingInfoCallback(int pingId, Promise<PingResultInfo> promise)
1373 {
1374 super(pingId);
1375 this.promise = promise;
1376 }
1377
1378 @Override
1379 public void succeeded()
1380 {
1381 if (promise != null)
1382 promise.succeeded(this);
1383 }
1384
1385 @Override
1386 public void failed(Throwable x)
1387 {
1388 if (promise != null)
1389 promise.failed(x);
1390 }
1391 }
1392 }