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