1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.websocket.common.io;
20
21 import java.io.EOFException;
22 import java.io.IOException;
23 import java.util.List;
24 import java.util.concurrent.CopyOnWriteArrayList;
25 import java.util.concurrent.atomic.AtomicReference;
26
27 import org.eclipse.jetty.util.StringUtil;
28 import org.eclipse.jetty.util.log.Log;
29 import org.eclipse.jetty.util.log.Logger;
30 import org.eclipse.jetty.websocket.api.StatusCode;
31 import org.eclipse.jetty.websocket.common.CloseInfo;
32 import org.eclipse.jetty.websocket.common.ConnectionState;
33
34
35
36
37
38
39
40
41
42 public class IOState
43 {
44
45
46
47 private static enum CloseHandshakeSource
48 {
49
50 NONE,
51
52 LOCAL,
53
54 REMOTE,
55
56 ABNORMAL;
57 }
58
59 public static interface ConnectionStateListener
60 {
61 public void onConnectionStateChange(ConnectionState state);
62 }
63
64 private static final Logger LOG = Log.getLogger(IOState.class);
65 private ConnectionState state;
66 private final List<ConnectionStateListener> listeners = new CopyOnWriteArrayList<>();
67
68
69
70
71
72 private boolean inputAvailable;
73
74
75
76
77 private boolean outputAvailable;
78
79
80
81
82 private CloseHandshakeSource closeHandshakeSource;
83
84
85
86
87
88
89 private CloseInfo closeInfo;
90
91
92
93
94
95 private AtomicReference<CloseInfo> finalClose = new AtomicReference<>();
96
97
98
99
100 private boolean cleanClose;
101
102
103
104
105 public IOState()
106 {
107 this.state = ConnectionState.CONNECTING;
108 this.inputAvailable = false;
109 this.outputAvailable = false;
110 this.closeHandshakeSource = CloseHandshakeSource.NONE;
111 this.closeInfo = null;
112 this.cleanClose = false;
113 }
114
115 public void addListener(ConnectionStateListener listener)
116 {
117 listeners.add(listener);
118 }
119
120 public void assertInputOpen() throws IOException
121 {
122 if (!isInputAvailable())
123 {
124 throw new IOException("Connection input is closed");
125 }
126 }
127
128 public void assertOutputOpen() throws IOException
129 {
130 if (!isOutputAvailable())
131 {
132 throw new IOException("Connection output is closed");
133 }
134 }
135
136 public CloseInfo getCloseInfo()
137 {
138 CloseInfo ci = finalClose.get();
139 if (ci != null)
140 {
141 return ci;
142 }
143 return closeInfo;
144 }
145
146 public ConnectionState getConnectionState()
147 {
148 return state;
149 }
150
151 public boolean isClosed()
152 {
153 synchronized (state)
154 {
155 return (state == ConnectionState.CLOSED);
156 }
157 }
158
159 public boolean isInputAvailable()
160 {
161 return inputAvailable;
162 }
163
164 public boolean isOpen()
165 {
166 return (getConnectionState() != ConnectionState.CLOSED);
167 }
168
169 public boolean isOutputAvailable()
170 {
171 return outputAvailable;
172 }
173
174 private void notifyStateListeners(ConnectionState state)
175 {
176 if (LOG.isDebugEnabled())
177 LOG.debug("Notify State Listeners: {}",state);
178 for (ConnectionStateListener listener : listeners)
179 {
180 if (LOG.isDebugEnabled())
181 {
182 LOG.debug("{}.onConnectionStateChange({})",listener.getClass().getSimpleName(),state.name());
183 }
184 listener.onConnectionStateChange(state);
185 }
186 }
187
188
189
190
191
192
193 public void onAbnormalClose(CloseInfo close)
194 {
195 if (LOG.isDebugEnabled())
196 LOG.debug("onAbnormalClose({})",close);
197 ConnectionState event = null;
198 synchronized (this)
199 {
200 if (this.state == ConnectionState.CLOSED)
201 {
202
203 return;
204 }
205
206 if (this.state == ConnectionState.OPEN)
207 {
208 this.cleanClose = false;
209 }
210
211 this.state = ConnectionState.CLOSED;
212 finalClose.compareAndSet(null,close);
213 this.inputAvailable = false;
214 this.outputAvailable = false;
215 this.closeHandshakeSource = CloseHandshakeSource.ABNORMAL;
216 event = this.state;
217 }
218 notifyStateListeners(event);
219 }
220
221
222
223
224 public void onCloseLocal(CloseInfo close)
225 {
226 ConnectionState event = null;
227 ConnectionState abnormalEvent = null;
228 ConnectionState initialState = this.state;
229 if (LOG.isDebugEnabled())
230 LOG.debug("onCloseLocal({}) : {}",close,initialState);
231 if (initialState == ConnectionState.CLOSED)
232 {
233
234 LOG.debug("already closed");
235 return;
236 }
237
238 if (initialState == ConnectionState.CONNECTED)
239 {
240
241 LOG.debug("FastClose in CONNECTED detected");
242
243 onOpened();
244 if (LOG.isDebugEnabled())
245 LOG.debug("FastClose continuing with Closure");
246 }
247
248 synchronized (this)
249 {
250 closeInfo = close;
251
252 boolean in = inputAvailable;
253 boolean out = outputAvailable;
254 if (closeHandshakeSource == CloseHandshakeSource.NONE)
255 {
256 closeHandshakeSource = CloseHandshakeSource.LOCAL;
257 }
258 out = false;
259 outputAvailable = false;
260
261 LOG.debug("onCloseLocal(), input={}, output={}",in,out);
262
263 if (!in && !out)
264 {
265 LOG.debug("Close Handshake satisfied, disconnecting");
266 cleanClose = true;
267 this.state = ConnectionState.CLOSED;
268 finalClose.compareAndSet(null,close);
269 event = this.state;
270 }
271 else if (this.state == ConnectionState.OPEN)
272 {
273
274 this.state = ConnectionState.CLOSING;
275 event = this.state;
276
277
278 if (close.isAbnormal())
279 {
280 abnormalEvent = ConnectionState.CLOSED;
281 finalClose.compareAndSet(null,close);
282 cleanClose = false;
283 outputAvailable = false;
284 inputAvailable = false;
285 closeHandshakeSource = CloseHandshakeSource.ABNORMAL;
286 }
287 }
288 }
289
290
291 if (event != null)
292 {
293 notifyStateListeners(event);
294
295 if(abnormalEvent != null)
296 {
297 notifyStateListeners(abnormalEvent);
298 }
299 }
300 }
301
302
303
304
305 public void onCloseRemote(CloseInfo close)
306 {
307 if (LOG.isDebugEnabled())
308 LOG.debug("onCloseRemote({})",close);
309 ConnectionState event = null;
310 synchronized (this)
311 {
312 if (this.state == ConnectionState.CLOSED)
313 {
314
315 return;
316 }
317
318 closeInfo = close;
319
320 boolean in = inputAvailable;
321 boolean out = outputAvailable;
322 if (closeHandshakeSource == CloseHandshakeSource.NONE)
323 {
324 closeHandshakeSource = CloseHandshakeSource.REMOTE;
325 }
326 in = false;
327 inputAvailable = false;
328
329 if (LOG.isDebugEnabled())
330 LOG.debug("onCloseRemote(), input={}, output={}",in,out);
331
332 if (!in && !out)
333 {
334 LOG.debug("Close Handshake satisfied, disconnecting");
335 cleanClose = true;
336 state = ConnectionState.CLOSED;
337 finalClose.compareAndSet(null,close);
338 event = this.state;
339 }
340 else if (this.state == ConnectionState.OPEN)
341 {
342
343 this.state = ConnectionState.CLOSING;
344 event = this.state;
345 }
346 }
347
348
349 if (event != null)
350 {
351 notifyStateListeners(event);
352 }
353 }
354
355
356
357
358
359
360 public void onConnected()
361 {
362 ConnectionState event = null;
363 synchronized (this)
364 {
365 if (this.state != ConnectionState.CONNECTING)
366 {
367 LOG.debug("Unable to set to connected, not in CONNECTING state: {}",this.state);
368 return;
369 }
370
371 this.state = ConnectionState.CONNECTED;
372 inputAvailable = false;
373 outputAvailable = true;
374 event = this.state;
375 }
376 notifyStateListeners(event);
377 }
378
379
380
381
382 public void onFailedUpgrade()
383 {
384 assert (this.state == ConnectionState.CONNECTING);
385 ConnectionState event = null;
386 synchronized (this)
387 {
388 this.state = ConnectionState.CLOSED;
389 cleanClose = false;
390 inputAvailable = false;
391 outputAvailable = false;
392 event = this.state;
393 }
394 notifyStateListeners(event);
395 }
396
397
398
399
400 public void onOpened()
401 {
402 ConnectionState event = null;
403 synchronized (this)
404 {
405 if (this.state == ConnectionState.OPEN)
406 {
407
408 return;
409 }
410
411 if (this.state != ConnectionState.CONNECTED)
412 {
413 LOG.debug("Unable to open, not in CONNECTED state: {}",this.state);
414 return;
415 }
416
417 this.state = ConnectionState.OPEN;
418 this.inputAvailable = true;
419 this.outputAvailable = true;
420 event = this.state;
421 }
422 notifyStateListeners(event);
423 }
424
425
426
427
428
429
430 public void onReadFailure(Throwable t)
431 {
432 ConnectionState event = null;
433 synchronized (this)
434 {
435 if (this.state == ConnectionState.CLOSED)
436 {
437
438 return;
439 }
440
441
442 String reason = "WebSocket Read Failure";
443 if (t instanceof EOFException)
444 {
445 reason = "WebSocket Read EOF";
446 Throwable cause = t.getCause();
447 if ((cause != null) && (StringUtil.isNotBlank(cause.getMessage())))
448 {
449 reason = "EOF: " + cause.getMessage();
450 }
451 }
452 else
453 {
454 if (StringUtil.isNotBlank(t.getMessage()))
455 {
456 reason = t.getMessage();
457 }
458 }
459
460 CloseInfo close = new CloseInfo(StatusCode.ABNORMAL,reason);
461
462 finalClose.compareAndSet(null,close);
463
464 this.cleanClose = false;
465 this.state = ConnectionState.CLOSED;
466 this.closeInfo = close;
467 this.inputAvailable = false;
468 this.outputAvailable = false;
469 this.closeHandshakeSource = CloseHandshakeSource.ABNORMAL;
470 event = this.state;
471 }
472 notifyStateListeners(event);
473 }
474
475
476
477
478
479
480 public void onWriteFailure(Throwable t)
481 {
482 ConnectionState event = null;
483 synchronized (this)
484 {
485 if (this.state == ConnectionState.CLOSED)
486 {
487
488 return;
489 }
490
491
492 String reason = "WebSocket Write Failure";
493 if (t instanceof EOFException)
494 {
495 reason = "WebSocket Write EOF";
496 Throwable cause = t.getCause();
497 if ((cause != null) && (StringUtil.isNotBlank(cause.getMessage())))
498 {
499 reason = "EOF: " + cause.getMessage();
500 }
501 }
502 else
503 {
504 if (StringUtil.isNotBlank(t.getMessage()))
505 {
506 reason = t.getMessage();
507 }
508 }
509
510 CloseInfo close = new CloseInfo(StatusCode.ABNORMAL,reason);
511
512 finalClose.compareAndSet(null,close);
513
514 this.cleanClose = false;
515 this.state = ConnectionState.CLOSED;
516 this.inputAvailable = false;
517 this.outputAvailable = false;
518 this.closeHandshakeSource = CloseHandshakeSource.ABNORMAL;
519 event = this.state;
520 }
521 notifyStateListeners(event);
522 }
523
524 public void onDisconnected()
525 {
526 ConnectionState event = null;
527 synchronized (this)
528 {
529 if (this.state == ConnectionState.CLOSED)
530 {
531
532 return;
533 }
534
535 CloseInfo close = new CloseInfo(StatusCode.ABNORMAL,"Disconnected");
536
537 this.cleanClose = false;
538 this.state = ConnectionState.CLOSED;
539 this.closeInfo = close;
540 this.inputAvailable = false;
541 this.outputAvailable = false;
542 this.closeHandshakeSource = CloseHandshakeSource.ABNORMAL;
543 event = this.state;
544 }
545 notifyStateListeners(event);
546 }
547
548 @Override
549 public String toString()
550 {
551 StringBuilder str = new StringBuilder();
552 str.append(this.getClass().getSimpleName());
553 str.append("@").append(Integer.toHexString(hashCode()));
554 str.append("[").append(state);
555 str.append(',');
556 if (!inputAvailable)
557 {
558 str.append('!');
559 }
560 str.append("in,");
561 if (!outputAvailable)
562 {
563 str.append('!');
564 }
565 str.append("out");
566 if ((state == ConnectionState.CLOSED) || (state == ConnectionState.CLOSING))
567 {
568 CloseInfo ci = finalClose.get();
569 if (ci != null)
570 {
571 str.append(",finalClose=").append(ci);
572 }
573 else
574 {
575 str.append(",close=").append(closeInfo);
576 }
577 str.append(",clean=").append(cleanClose);
578 str.append(",closeSource=").append(closeHandshakeSource);
579 }
580 str.append(']');
581 return str.toString();
582 }
583
584 public boolean wasAbnormalClose()
585 {
586 return closeHandshakeSource == CloseHandshakeSource.ABNORMAL;
587 }
588
589 public boolean wasCleanClose()
590 {
591 return cleanClose;
592 }
593
594 public boolean wasLocalCloseInitiated()
595 {
596 return closeHandshakeSource == CloseHandshakeSource.LOCAL;
597 }
598
599 public boolean wasRemoteCloseInitiated()
600 {
601 return closeHandshakeSource == CloseHandshakeSource.REMOTE;
602 }
603
604 }