1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
package org.eclipse.jetty.util;
import java.io.IOException;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
public class FutureCallback<C> implements Future<C>,Callback<C>
{
// TODO investigate use of a phasor
private enum State {NOT_DONE,DOING,DONE};
private final AtomicReference<State> _state=new AtomicReference<>(State.NOT_DONE);
private CountDownLatch _done= new CountDownLatch(1);
private Throwable _cause;
private C _context;
private boolean _completed;
private void recycle()
{
// TODO make this public?
if (!isDone())
throw new IllegalStateException();
_cause=null;
_context=null;
_completed=false;
_done=new CountDownLatch(1);
}
@Override
public void completed(C context)
{
if (_state.compareAndSet(State.NOT_DONE,State.DOING))
{
_context=context;
_completed=true;
if (_state.compareAndSet(State.DOING,State.DONE))
{
_done.countDown();
return;
}
}
else if (!isCancelled())
throw new IllegalStateException();
}
@Override
public void failed(C context, Throwable cause)
{
if (_state.compareAndSet(State.NOT_DONE,State.DOING))
{
_context=context;
_cause=cause;
if (_state.compareAndSet(State.DOING,State.DONE))
{
_done.countDown();
return;
}
}
else if (!isCancelled())
throw new IllegalStateException();
}
@Override
public boolean cancel(boolean mayInterruptIfRunning)
{
failed(null,new CancellationException());
return false;
}
@Override
public boolean isCancelled()
{
return State.DONE.equals(_state.get())&&_cause instanceof CancellationException;
}
@Override
public boolean isDone()
{
return State.DONE.equals(_state.get());
}
@Override
public C get() throws InterruptedException, ExecutionException
{
_done.await();
if (_completed)
return _context;
if (_cause instanceof CancellationException)
throw (CancellationException) new CancellationException().initCause(_cause);
throw new ExecutionException(_cause);
}
@Override
public C get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
{
if (!_done.await(timeout,unit))
throw new TimeoutException();
if (_completed)
return _context;
if (_cause instanceof TimeoutException)
throw (TimeoutException)_cause;
if (_cause instanceof CancellationException)
throw (CancellationException) new CancellationException().initCause(_cause);
throw new ExecutionException(_cause);
}
public static void rethrow(ExecutionException e) throws IOException
{
Throwable cause=e.getCause();
if (cause instanceof IOException)
throw (IOException)cause;
if (cause instanceof Error)
throw (Error)cause;
if (cause instanceof RuntimeException)
throw (RuntimeException)cause;
throw new RuntimeException(cause);
}
/* ------------------------------------------------------------ */
public String toString()
{
return String.format("FutureCallback@%x{%s,%b,%s}",hashCode(),_state,_completed,_context);
}
}
|
