1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.http2;
20
21 import java.io.EOFException;
22 import java.io.IOException;
23 import java.util.concurrent.ConcurrentHashMap;
24 import java.util.concurrent.ConcurrentMap;
25 import java.util.concurrent.TimeoutException;
26 import java.util.concurrent.atomic.AtomicInteger;
27 import java.util.concurrent.atomic.AtomicReference;
28
29 import org.eclipse.jetty.http2.api.Stream;
30 import org.eclipse.jetty.http2.frames.DataFrame;
31 import org.eclipse.jetty.http2.frames.Frame;
32 import org.eclipse.jetty.http2.frames.HeadersFrame;
33 import org.eclipse.jetty.http2.frames.PushPromiseFrame;
34 import org.eclipse.jetty.http2.frames.ResetFrame;
35 import org.eclipse.jetty.http2.frames.WindowUpdateFrame;
36 import org.eclipse.jetty.io.IdleTimeout;
37 import org.eclipse.jetty.util.Callback;
38 import org.eclipse.jetty.util.Promise;
39 import org.eclipse.jetty.util.log.Log;
40 import org.eclipse.jetty.util.log.Logger;
41 import org.eclipse.jetty.util.thread.Scheduler;
42
43 public class HTTP2Stream extends IdleTimeout implements IStream
44 {
45 private static final Logger LOG = Log.getLogger(HTTP2Stream.class);
46
47 private final AtomicReference<ConcurrentMap<String, Object>> attributes = new AtomicReference<>();
48 private final AtomicReference<CloseState> closeState = new AtomicReference<>(CloseState.NOT_CLOSED);
49 private final AtomicInteger sendWindow = new AtomicInteger();
50 private final AtomicInteger recvWindow = new AtomicInteger();
51 private final ISession session;
52 private final int streamId;
53 private final boolean local;
54 private volatile Listener listener;
55 private volatile boolean localReset;
56 private volatile boolean remoteReset;
57
58 public HTTP2Stream(Scheduler scheduler, ISession session, int streamId, boolean local)
59 {
60 super(scheduler);
61 this.session = session;
62 this.streamId = streamId;
63 this.local = local;
64 }
65
66 @Override
67 public int getId()
68 {
69 return streamId;
70 }
71
72 @Override
73 public boolean isLocal()
74 {
75 return local;
76 }
77
78 @Override
79 public ISession getSession()
80 {
81 return session;
82 }
83
84 @Override
85 public void headers(HeadersFrame frame, Callback callback)
86 {
87 session.frames(this, callback, frame, Frame.EMPTY_ARRAY);
88 }
89
90 @Override
91 public void push(PushPromiseFrame frame, Promise<Stream> promise, Listener listener)
92 {
93 session.push(this, promise, frame, listener);
94 }
95
96 @Override
97 public void data(DataFrame frame, Callback callback)
98 {
99 session.data(this, callback, frame);
100 }
101
102 @Override
103 public void reset(ResetFrame frame, Callback callback)
104 {
105 if (isReset())
106 return;
107 localReset = true;
108 session.frames(this, callback, frame, Frame.EMPTY_ARRAY);
109 }
110
111 @Override
112 public Object getAttribute(String key)
113 {
114 return attributes().get(key);
115 }
116
117 @Override
118 public void setAttribute(String key, Object value)
119 {
120 attributes().put(key, value);
121 }
122
123 @Override
124 public Object removeAttribute(String key)
125 {
126 return attributes().remove(key);
127 }
128
129 @Override
130 public boolean isReset()
131 {
132 return localReset || remoteReset;
133 }
134
135 @Override
136 public boolean isClosed()
137 {
138 return closeState.get() == CloseState.CLOSED;
139 }
140
141 public boolean isRemotelyClosed()
142 {
143 return closeState.get() == CloseState.REMOTELY_CLOSED;
144 }
145
146 public boolean isLocallyClosed()
147 {
148 return closeState.get() == CloseState.LOCALLY_CLOSED;
149 }
150
151 @Override
152 public boolean isOpen()
153 {
154 return !isClosed();
155 }
156
157 @Override
158 protected void onIdleExpired(TimeoutException timeout)
159 {
160 if (LOG.isDebugEnabled())
161 LOG.debug("Idle timeout {}ms expired on {}", getIdleTimeout(), this);
162
163
164 if (notifyIdleTimeout(this, timeout))
165 {
166
167 reset(new ResetFrame(getId(), ErrorCode.CANCEL_STREAM_ERROR.code), Callback.NOOP);
168 }
169 }
170
171 private ConcurrentMap<String, Object> attributes()
172 {
173 ConcurrentMap<String, Object> map = attributes.get();
174 if (map == null)
175 {
176 map = new ConcurrentHashMap<>();
177 if (!attributes.compareAndSet(null, map))
178 {
179 map = attributes.get();
180 }
181 }
182 return map;
183 }
184
185 @Override
186 public Listener getListener()
187 {
188 return listener;
189 }
190
191 @Override
192 public void setListener(Listener listener)
193 {
194 this.listener = listener;
195 }
196
197 @Override
198 public void process(Frame frame, Callback callback)
199 {
200 notIdle();
201 switch (frame.getType())
202 {
203 case HEADERS:
204 {
205 onHeaders((HeadersFrame)frame, callback);
206 break;
207 }
208 case DATA:
209 {
210 onData((DataFrame)frame, callback);
211 break;
212 }
213 case RST_STREAM:
214 {
215 onReset((ResetFrame)frame, callback);
216 break;
217 }
218 case PUSH_PROMISE:
219 {
220 onPush((PushPromiseFrame)frame, callback);
221 break;
222 }
223 case WINDOW_UPDATE:
224 {
225 onWindowUpdate((WindowUpdateFrame)frame, callback);
226 break;
227 }
228 default:
229 {
230 throw new UnsupportedOperationException();
231 }
232 }
233 }
234
235 private void onHeaders(HeadersFrame frame, Callback callback)
236 {
237 if (updateClose(frame.isEndStream(), false))
238 session.removeStream(this);
239 callback.succeeded();
240 }
241
242 private void onData(DataFrame frame, Callback callback)
243 {
244 if (getRecvWindow() < 0)
245 {
246
247
248 session.close(ErrorCode.FLOW_CONTROL_ERROR.code, "stream_window_exceeded", Callback.NOOP);
249 callback.failed(new IOException("stream_window_exceeded"));
250 return;
251 }
252
253
254 if (isRemotelyClosed())
255 {
256 reset(new ResetFrame(streamId, ErrorCode.STREAM_CLOSED_ERROR.code), Callback.NOOP);
257 callback.failed(new EOFException("stream_closed"));
258 return;
259 }
260
261 if (isReset())
262 {
263
264 callback.failed(new IOException("stream_reset"));
265 return;
266 }
267
268 if (updateClose(frame.isEndStream(), false))
269 session.removeStream(this);
270 notifyData(this, frame, callback);
271 }
272
273 private void onReset(ResetFrame frame, Callback callback)
274 {
275 remoteReset = true;
276 close();
277 session.removeStream(this);
278 callback.succeeded();
279 notifyReset(this, frame);
280 }
281
282 private void onPush(PushPromiseFrame frame, Callback callback)
283 {
284
285
286 updateClose(true, true);
287 callback.succeeded();
288 }
289
290 private void onWindowUpdate(WindowUpdateFrame frame, Callback callback)
291 {
292 callback.succeeded();
293 }
294
295 @Override
296 public boolean updateClose(boolean update, boolean local)
297 {
298 if (LOG.isDebugEnabled())
299 LOG.debug("Update close for {} close={} local={}", this, update, local);
300
301 if (!update)
302 return false;
303
304 while (true)
305 {
306 CloseState current = closeState.get();
307 switch (current)
308 {
309 case NOT_CLOSED:
310 {
311 CloseState newValue = local ? CloseState.LOCALLY_CLOSED : CloseState.REMOTELY_CLOSED;
312 if (closeState.compareAndSet(current, newValue))
313 return false;
314 break;
315 }
316 case LOCALLY_CLOSED:
317 {
318 if (local)
319 return false;
320 close();
321 return true;
322 }
323 case REMOTELY_CLOSED:
324 {
325 if (!local)
326 return false;
327 close();
328 return true;
329 }
330 default:
331 {
332 return false;
333 }
334 }
335 }
336 }
337
338 public int getSendWindow()
339 {
340 return sendWindow.get();
341 }
342
343 public int getRecvWindow()
344 {
345 return recvWindow.get();
346 }
347
348 @Override
349 public int updateSendWindow(int delta)
350 {
351 return sendWindow.getAndAdd(delta);
352 }
353
354 @Override
355 public int updateRecvWindow(int delta)
356 {
357 return recvWindow.getAndAdd(delta);
358 }
359
360 @Override
361 public void close()
362 {
363 closeState.set(CloseState.CLOSED);
364 onClose();
365 }
366
367 private void notifyData(Stream stream, DataFrame frame, Callback callback)
368 {
369 final Listener listener = this.listener;
370 if (listener == null)
371 return;
372 try
373 {
374 listener.onData(stream, frame, callback);
375 }
376 catch (Throwable x)
377 {
378 LOG.info("Failure while notifying listener " + listener, x);
379 }
380 }
381
382 private void notifyReset(Stream stream, ResetFrame frame)
383 {
384 final Listener listener = this.listener;
385 if (listener == null)
386 return;
387 try
388 {
389 listener.onReset(stream, frame);
390 }
391 catch (Throwable x)
392 {
393 LOG.info("Failure while notifying listener " + listener, x);
394 }
395 }
396
397 private boolean notifyIdleTimeout(Stream stream, Throwable failure)
398 {
399 Listener listener = this.listener;
400 if (listener == null)
401 return true;
402 try
403 {
404 return listener.onIdleTimeout(stream, failure);
405 }
406 catch (Throwable x)
407 {
408 LOG.info("Failure while notifying listener " + listener, x);
409 return true;
410 }
411 }
412
413 @Override
414 public String toString()
415 {
416 return String.format("%s@%x#%d{sendWindow=%s,recvWindow=%s,reset=%b,%s}", getClass().getSimpleName(),
417 hashCode(), getId(), sendWindow, recvWindow, isReset(), closeState);
418 }
419 }