1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.http2.client;
20
21 import java.util.concurrent.atomic.AtomicLong;
22
23 import org.eclipse.jetty.http2.FlowControlStrategy;
24 import org.eclipse.jetty.http2.HTTP2Session;
25 import org.eclipse.jetty.http2.IStream;
26 import org.eclipse.jetty.http2.api.Session;
27 import org.eclipse.jetty.http2.api.Stream;
28 import org.eclipse.jetty.http2.frames.HeadersFrame;
29 import org.eclipse.jetty.http2.frames.PushPromiseFrame;
30 import org.eclipse.jetty.http2.generator.Generator;
31 import org.eclipse.jetty.io.EndPoint;
32 import org.eclipse.jetty.util.Callback;
33 import org.eclipse.jetty.util.log.Log;
34 import org.eclipse.jetty.util.log.Logger;
35 import org.eclipse.jetty.util.thread.Scheduler;
36
37 public class HTTP2ClientSession extends HTTP2Session
38 {
39 private static final Logger LOG = Log.getLogger(HTTP2ClientSession.class);
40
41 private final AtomicLong streamsOpened = new AtomicLong();
42 private final AtomicLong streamsClosed = new AtomicLong();
43
44 public HTTP2ClientSession(Scheduler scheduler, EndPoint endPoint, Generator generator, Session.Listener listener, FlowControlStrategy flowControl)
45 {
46 super(scheduler, endPoint, generator, listener, flowControl, 1);
47 }
48
49 @Override
50 protected void onStreamOpened(IStream stream)
51 {
52 super.onStreamOpened(stream);
53 streamsOpened.incrementAndGet();
54 }
55
56 @Override
57 protected void onStreamClosed(IStream stream)
58 {
59 super.onStreamClosed(stream);
60 streamsClosed.incrementAndGet();
61 }
62
63 public long getStreamsOpened()
64 {
65 return streamsOpened.get();
66 }
67
68 public long getStreamsClosed()
69 {
70 return streamsClosed.get();
71 }
72
73 @Override
74 public void onHeaders(HeadersFrame frame)
75 {
76 if (LOG.isDebugEnabled())
77 LOG.debug("Received {}", frame);
78
79 int streamId = frame.getStreamId();
80 IStream stream = getStream(streamId);
81 if (stream == null)
82 {
83 if (LOG.isDebugEnabled())
84 LOG.debug("Ignoring {}, stream #{} not found", frame, streamId);
85 }
86 else
87 {
88 stream.process(frame, Callback.NOOP);
89 notifyHeaders(stream, frame);
90 }
91 }
92
93 private void notifyHeaders(IStream stream, HeadersFrame frame)
94 {
95 Stream.Listener listener = stream.getListener();
96 if (listener == null)
97 return;
98 try
99 {
100 listener.onHeaders(stream, frame);
101 }
102 catch (Throwable x)
103 {
104 LOG.info("Failure while notifying listener " + listener, x);
105 }
106 }
107
108 @Override
109 public void onPushPromise(PushPromiseFrame frame)
110 {
111 if (LOG.isDebugEnabled())
112 LOG.debug("Received {}", frame);
113
114 int streamId = frame.getStreamId();
115 int pushStreamId = frame.getPromisedStreamId();
116 IStream stream = getStream(streamId);
117 if (stream == null)
118 {
119 if (LOG.isDebugEnabled())
120 LOG.debug("Ignoring {}, stream #{} not found", frame, streamId);
121 }
122 else
123 {
124 IStream pushStream = createRemoteStream(pushStreamId);
125 pushStream.process(frame, Callback.NOOP);
126 Stream.Listener listener = notifyPush(stream, pushStream, frame);
127 pushStream.setListener(listener);
128 }
129 }
130
131 private Stream.Listener notifyPush(IStream stream, IStream pushStream, PushPromiseFrame frame)
132 {
133 Stream.Listener listener = stream.getListener();
134 if (listener == null)
135 return null;
136 try
137 {
138 return listener.onPush(pushStream, frame);
139 }
140 catch (Throwable x)
141 {
142 LOG.info("Failure while notifying listener " + listener, x);
143 return null;
144 }
145 }
146 }