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.util.Collections;
22 import java.util.Map;
23 import java.util.Set;
24 import java.util.concurrent.ConcurrentHashMap;
25 import java.util.concurrent.Future;
26 import java.util.concurrent.TimeUnit;
27 import java.util.concurrent.atomic.AtomicInteger;
28
29 import org.eclipse.jetty.spdy.api.DataInfo;
30 import org.eclipse.jetty.spdy.api.HeadersInfo;
31 import org.eclipse.jetty.spdy.api.ReplyInfo;
32 import org.eclipse.jetty.spdy.api.RstInfo;
33 import org.eclipse.jetty.spdy.api.Stream;
34 import org.eclipse.jetty.spdy.api.StreamFrameListener;
35 import org.eclipse.jetty.spdy.api.StreamStatus;
36 import org.eclipse.jetty.spdy.api.SynInfo;
37 import org.eclipse.jetty.spdy.frames.ControlFrame;
38 import org.eclipse.jetty.spdy.frames.HeadersFrame;
39 import org.eclipse.jetty.spdy.frames.SynReplyFrame;
40 import org.eclipse.jetty.util.Callback;
41 import org.eclipse.jetty.util.FutureCallback;
42 import org.eclipse.jetty.util.log.Log;
43 import org.eclipse.jetty.util.log.Logger;
44
45 public class StandardStream implements IStream
46 {
47 private static final Logger LOG = Log.getLogger(Stream.class);
48 private final Map<String, Object> attributes = new ConcurrentHashMap<>();
49 private final int id;
50 private final byte priority;
51 private final ISession session;
52 private final IStream associatedStream;
53 private final AtomicInteger windowSize = new AtomicInteger();
54 private final Set<Stream> pushedStreams = Collections.newSetFromMap(new ConcurrentHashMap<Stream, Boolean>());
55 private volatile StreamFrameListener listener;
56 private volatile OpenState openState = OpenState.SYN_SENT;
57 private volatile CloseState closeState = CloseState.OPENED;
58 private volatile boolean reset = false;
59
60 public StandardStream(int id, byte priority, ISession session, IStream associatedStream)
61 {
62 this.id = id;
63 this.priority = priority;
64 this.session = session;
65 this.associatedStream = associatedStream;
66 }
67
68 @Override
69 public int getId()
70 {
71 return id;
72 }
73
74 @Override
75 public IStream getAssociatedStream()
76 {
77 return associatedStream;
78 }
79
80 @Override
81 public Set<Stream> getPushedStreams()
82 {
83 return pushedStreams;
84 }
85
86 @Override
87 public void associate(IStream stream)
88 {
89 pushedStreams.add(stream);
90 }
91
92 @Override
93 public void disassociate(IStream stream)
94 {
95 pushedStreams.remove(stream);
96 }
97
98 @Override
99 public byte getPriority()
100 {
101 return priority;
102 }
103
104 @Override
105 public int getWindowSize()
106 {
107 return windowSize.get();
108 }
109
110 @Override
111 public void updateWindowSize(int delta)
112 {
113 int size = windowSize.addAndGet(delta);
114 LOG.debug("Updated window size {} -> {} for {}", size - delta, size, this);
115 }
116
117 @Override
118 public ISession getSession()
119 {
120 return session;
121 }
122
123 @Override
124 public Object getAttribute(String key)
125 {
126 return attributes.get(key);
127 }
128
129 @Override
130 public void setAttribute(String key, Object value)
131 {
132 attributes.put(key,value);
133 }
134
135 @Override
136 public Object removeAttribute(String key)
137 {
138 return attributes.remove(key);
139 }
140
141 @Override
142 public void setStreamFrameListener(StreamFrameListener listener)
143 {
144 this.listener = listener;
145 }
146
147 public StreamFrameListener getStreamFrameListener()
148 {
149 return listener;
150 }
151
152 @Override
153 public void updateCloseState(boolean close, boolean local)
154 {
155 LOG.debug("{} close={} local={}",this,close,local);
156 if (close)
157 {
158 switch (closeState)
159 {
160 case OPENED:
161 {
162 closeState = local ? CloseState.LOCALLY_CLOSED : CloseState.REMOTELY_CLOSED;
163 break;
164 }
165 case LOCALLY_CLOSED:
166 {
167 if (local)
168 throw new IllegalStateException();
169 else
170 closeState = CloseState.CLOSED;
171 break;
172 }
173 case REMOTELY_CLOSED:
174 {
175 if (local)
176 closeState = CloseState.CLOSED;
177 else
178 throw new IllegalStateException();
179 break;
180 }
181 default:
182 {
183 LOG.warn("Already CLOSED! {} local={}",this,local);
184 }
185 }
186 }
187 }
188
189 @Override
190 public void process(ControlFrame frame)
191 {
192 switch (frame.getType())
193 {
194 case SYN_STREAM:
195 {
196 openState = OpenState.SYN_RECV;
197 break;
198 }
199 case SYN_REPLY:
200 {
201 openState = OpenState.REPLY_RECV;
202 SynReplyFrame synReply = (SynReplyFrame)frame;
203 updateCloseState(synReply.isClose(), false);
204 ReplyInfo replyInfo = new ReplyInfo(synReply.getHeaders(), synReply.isClose());
205 notifyOnReply(replyInfo);
206 break;
207 }
208 case HEADERS:
209 {
210 HeadersFrame headers = (HeadersFrame)frame;
211 updateCloseState(headers.isClose(), false);
212 HeadersInfo headersInfo = new HeadersInfo(headers.getHeaders(), headers.isClose(), headers.isResetCompression());
213 notifyOnHeaders(headersInfo);
214 break;
215 }
216 case RST_STREAM:
217 {
218 reset = true;
219 break;
220 }
221 default:
222 {
223 throw new IllegalStateException();
224 }
225 }
226 session.flush();
227 }
228
229 @Override
230 public void process(DataInfo dataInfo)
231 {
232
233
234 if (isRemotelyClosed())
235 {
236 LOG.debug("Stream is remotely closed, ignoring {}", dataInfo);
237 return;
238 }
239
240 if (!canReceive())
241 {
242 LOG.debug("Protocol error receiving {}, resetting" + dataInfo);
243 session.rst(new RstInfo(getId(), StreamStatus.PROTOCOL_ERROR));
244 return;
245 }
246
247 updateCloseState(dataInfo.isClose(), false);
248 notifyOnData(dataInfo);
249 session.flush();
250 }
251
252 private void notifyOnReply(ReplyInfo replyInfo)
253 {
254 final StreamFrameListener listener = this.listener;
255 try
256 {
257 if (listener != null)
258 {
259 LOG.debug("Invoking reply callback with {} on listener {}", replyInfo, listener);
260 listener.onReply(this, replyInfo);
261 }
262 }
263 catch (Exception x)
264 {
265 LOG.info("Exception while notifying listener " + listener, x);
266 }
267 catch (Error x)
268 {
269 LOG.info("Exception while notifying listener " + listener, x);
270 throw x;
271 }
272 }
273
274 private void notifyOnHeaders(HeadersInfo headersInfo)
275 {
276 final StreamFrameListener listener = this.listener;
277 try
278 {
279 if (listener != null)
280 {
281 LOG.debug("Invoking headers callback with {} on listener {}", headersInfo, listener);
282 listener.onHeaders(this, headersInfo);
283 }
284 }
285 catch (Exception x)
286 {
287 LOG.info("Exception while notifying listener " + listener, x);
288 }
289 catch (Error x)
290 {
291 LOG.info("Exception while notifying listener " + listener, x);
292 throw x;
293 }
294 }
295
296 private void notifyOnData(DataInfo dataInfo)
297 {
298 final StreamFrameListener listener = this.listener;
299 try
300 {
301 if (listener != null)
302 {
303 LOG.debug("Invoking data callback with {} on listener {}", dataInfo, listener);
304 listener.onData(this, dataInfo);
305 LOG.debug("Invoked data callback with {} on listener {}", dataInfo, listener);
306 }
307 }
308 catch (Exception x)
309 {
310 LOG.info("Exception while notifying listener " + listener, x);
311 }
312 catch (Error x)
313 {
314 LOG.info("Exception while notifying listener " + listener, x);
315 throw x;
316 }
317 }
318
319 @Override
320 public Future<Stream> syn(SynInfo synInfo)
321 {
322 Promise<Stream> result = new Promise<>();
323 syn(synInfo,0,TimeUnit.MILLISECONDS,result);
324 return result;
325 }
326
327 @Override
328 public void syn(SynInfo synInfo, long timeout, TimeUnit unit, Callback<Stream> callback)
329 {
330 if (isClosed() || isReset())
331 {
332 callback.failed(this, new StreamException(getId(), StreamStatus.STREAM_ALREADY_CLOSED,
333 "Stream: " + this + " already closed or reset!"));
334 return;
335 }
336 PushSynInfo pushSynInfo = new PushSynInfo(getId(), synInfo);
337 session.syn(pushSynInfo, null, timeout, unit, callback);
338 }
339
340 @Override
341 public Future<Void> reply(ReplyInfo replyInfo)
342 {
343 Promise<Void> result = new Promise<>();
344 reply(replyInfo,0,TimeUnit.MILLISECONDS,result);
345 return result;
346 }
347
348 @Override
349 public void reply(ReplyInfo replyInfo, long timeout, TimeUnit unit, Callback<Void> callback)
350 {
351 if (isUnidirectional())
352 throw new IllegalStateException("Protocol violation: cannot send SYN_REPLY frames in unidirectional streams");
353 openState = OpenState.REPLY_SENT;
354 updateCloseState(replyInfo.isClose(), true);
355 SynReplyFrame frame = new SynReplyFrame(session.getVersion(), replyInfo.getFlags(), getId(), replyInfo.getHeaders());
356 session.control(this, frame, timeout, unit, callback, null);
357 }
358
359 @Override
360 public Future<Void> data(DataInfo dataInfo)
361 {
362 FutureCallback<Void> fcb = new FutureCallback<>();
363 data(dataInfo,0,TimeUnit.MILLISECONDS,null,fcb);
364 return fcb;
365 }
366
367 @Override
368 public <C> void data(DataInfo dataInfo, long timeout, TimeUnit unit, C context, Callback<C> callback)
369 {
370 if (!canSend())
371 {
372 session.rst(new RstInfo(getId(), StreamStatus.PROTOCOL_ERROR));
373 throw new IllegalStateException("Protocol violation: cannot send a DATA frame before a SYN_REPLY frame");
374 }
375 if (isLocallyClosed())
376 {
377 session.rst(new RstInfo(getId(), StreamStatus.PROTOCOL_ERROR));
378 throw new IllegalStateException("Protocol violation: cannot send a DATA frame on a closed stream");
379 }
380
381
382
383 session.data(this, dataInfo, timeout, unit, callback, context);
384 }
385
386 @Override
387 public Future<Void> headers(HeadersInfo headersInfo)
388 {
389 FutureCallback<Void> fcb = new FutureCallback<>();
390 headers(headersInfo,0,TimeUnit.MILLISECONDS,fcb);
391 return fcb;
392 }
393
394 @Override
395 public void headers(HeadersInfo headersInfo, long timeout, TimeUnit unit, Callback<Void> callback)
396 {
397 if (!canSend())
398 {
399 session.rst(new RstInfo(getId(), StreamStatus.PROTOCOL_ERROR));
400 throw new IllegalStateException("Protocol violation: cannot send a HEADERS frame before a SYN_REPLY frame");
401 }
402 if (isLocallyClosed())
403 {
404 session.rst(new RstInfo(getId(), StreamStatus.PROTOCOL_ERROR));
405 throw new IllegalStateException("Protocol violation: cannot send a HEADERS frame on a closed stream");
406 }
407
408 updateCloseState(headersInfo.isClose(), true);
409 HeadersFrame frame = new HeadersFrame(session.getVersion(), headersInfo.getFlags(), getId(), headersInfo.getHeaders());
410 session.control(this, frame, timeout, unit, callback, null);
411 }
412
413 @Override
414 public boolean isUnidirectional()
415 {
416 return associatedStream != null;
417 }
418
419 @Override
420 public boolean isReset()
421 {
422 return reset;
423 }
424
425 @Override
426 public boolean isHalfClosed()
427 {
428 CloseState closeState = this.closeState;
429 return closeState == CloseState.LOCALLY_CLOSED || closeState == CloseState.REMOTELY_CLOSED || closeState == CloseState.CLOSED;
430 }
431
432 @Override
433 public boolean isClosed()
434 {
435 return closeState == CloseState.CLOSED;
436 }
437
438 private boolean isLocallyClosed()
439 {
440 CloseState closeState = this.closeState;
441 return closeState == CloseState.LOCALLY_CLOSED || closeState == CloseState.CLOSED;
442 }
443
444 private boolean isRemotelyClosed()
445 {
446 CloseState closeState = this.closeState;
447 return closeState == CloseState.REMOTELY_CLOSED || closeState == CloseState.CLOSED;
448 }
449
450 @Override
451 public String toString()
452 {
453 return String.format("stream=%d v%d windowSize=%db reset=%s %s %s", getId(), session.getVersion(), getWindowSize(), isReset(), openState, closeState);
454 }
455
456 private boolean canSend()
457 {
458 OpenState openState = this.openState;
459 return openState == OpenState.SYN_SENT || openState == OpenState.REPLY_RECV || openState == OpenState.REPLY_SENT;
460 }
461
462 private boolean canReceive()
463 {
464 OpenState openState = this.openState;
465 return openState == OpenState.SYN_RECV || openState == OpenState.REPLY_RECV || openState == OpenState.REPLY_SENT;
466 }
467
468 private enum OpenState
469 {
470 SYN_SENT, SYN_RECV, REPLY_SENT, REPLY_RECV
471 }
472
473 private enum CloseState
474 {
475 OPENED, LOCALLY_CLOSED, REMOTELY_CLOSED, CLOSED
476 }
477 }