Skip to content

Commit 54aa882

Browse files
koushkoush
committed
add support for futures on AsyncHttpClient. TODO: Futures on sockets?
1 parent 0249f29 commit 54aa882

18 files changed

Lines changed: 651 additions & 335 deletions

‎AndroidAsync/src/com/koushikdutta/async/AsyncServer.java‎

Lines changed: 332 additions & 103 deletions
Large diffs are not rendered by default.

‎AndroidAsync/src/com/koushikdutta/async/Cancelable.java‎

Lines changed: 0 additions & 7 deletions
This file was deleted.

‎AndroidAsync/src/com/koushikdutta/async/SimpleCancelable.java‎

Lines changed: 0 additions & 31 deletions
This file was deleted.

‎AndroidAsync/src/com/koushikdutta/async/callback/ContinuationCallback.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package com.koushikdutta.async.callback;
22

3-
import com.koushikdutta.async.Continuation;
3+
import com.koushikdutta.async.future.Continuation;
44

55
public interface ContinuationCallback {
66
public void onContinue(Continuation continuation, CompletedCallback next) throws Exception;

‎AndroidAsync/src/com/koushikdutta/async/future/Cancellable.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package com.koushikdutta.async.future;
2+
3+
public interface Cancellable {
4+
boolean isDone();
5+
boolean isCancelled();
6+
boolean cancel();
7+
}

AndroidAsync/src/com/koushikdutta/async/Continuation.java renamed to AndroidAsync/src/com/koushikdutta/async/future/Continuation.java

Lines changed: 25 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package com.koushikdutta.async;
1+
package com.koushikdutta.async.future;
22

33
import java.util.LinkedList;
44

@@ -7,7 +7,7 @@
77
import com.koushikdutta.async.callback.CompletedCallback;
88
import com.koushikdutta.async.callback.ContinuationCallback;
99

10-
public class Continuation implements ContinuationCallback, Runnable, Cancelable {
10+
public class Continuation extends SimpleCancelable implements ContinuationCallback, Runnable, Cancellable {
1111
CompletedCallback callback;
1212
Runnable cancelCallback;
1313

@@ -24,7 +24,7 @@ public Runnable getCancelCallback() {
2424
public void setCancelCallback(Runnable cancelCallback) {
2525
this.cancelCallback = cancelCallback;
2626
}
27-
public void setCancelCallback(final Cancelable cancel) {
27+
public void setCancelCallback(final Cancellable cancel) {
2828
if (cancel == null) {
2929
this.cancelCallback = null;
3030
return;
@@ -69,35 +69,37 @@ public void onCompleted(Exception ex) {
6969
};
7070
}
7171

72-
boolean completed;
7372
void reportCompleted(Exception ex) {
74-
if (cancel)
73+
if (!setComplete())
7574
return;
76-
completed = true;
7775
if (callback != null)
7876
callback.onCompleted(ex);
7977
}
8078

8179
LinkedList<ContinuationCallback> mCallbacks = new LinkedList<ContinuationCallback>();
8280

81+
private ContinuationCallback hook(ContinuationCallback callback) {
82+
if (callback instanceof SimpleCancelable) {
83+
SimpleCancelable child = (SimpleCancelable)callback;
84+
child.setParent(this);
85+
}
86+
return callback;
87+
}
88+
8389
public void add(ContinuationCallback callback) {
84-
mCallbacks.add(callback);
90+
mCallbacks.add(hook(callback));
8591
}
8692

8793
public void insert(ContinuationCallback callback) {
88-
mCallbacks.add(0, callback);
94+
mCallbacks.add(0, hook(callback));
8995
}
9096

9197
private boolean inNext;
9298
private boolean waiting;
9399
private void next() {
94100
if (inNext)
95101
return;
96-
if (isCanceled()) {
97-
reportCompleted(null);
98-
return;
99-
}
100-
while (mCallbacks.size() > 0 && !waiting && !completed && !cancel) {
102+
while (mCallbacks.size() > 0 && !waiting && !isDone() && !isCancelled()) {
101103
ContinuationCallback cb = mCallbacks.remove();
102104
try {
103105
inNext = true;
@@ -113,38 +115,25 @@ private void next() {
113115
}
114116
if (waiting)
115117
return;
116-
if (completed)
118+
if (isDone())
117119
return;
118-
if (cancel)
120+
if (isCancelled())
119121
return;
120122

121123
reportCompleted(null);
122124
}
123-
124-
public boolean isCompleted() {
125-
return completed;
126-
}
127-
128-
public boolean isCanceled() {
129-
return cancel || (parent != null && parent.isCanceled());
130-
}
131-
132-
public Cancelable cancel() {
133-
cancelThis();
134-
if (parent != null)
135-
parent.cancel();
136-
return this;
137-
}
138-
139-
public void cancelThis() {
140-
if (isCanceled())
141-
return;
142-
cancel = true;
125+
126+
@Override
127+
public boolean cancel() {
128+
if (!super.cancel())
129+
return false;
130+
143131
if (cancelCallback != null)
144132
cancelCallback.run();
133+
134+
return true;
145135
}
146136

147-
boolean cancel;
148137
boolean started;
149138
public Continuation start() {
150139
Assert.assertTrue(!started);
@@ -153,12 +142,9 @@ public Continuation start() {
153142
return this;
154143
}
155144

156-
Continuation parent;
157145
@Override
158146
public void onContinue(Continuation continuation, CompletedCallback next) throws Exception {
159-
parent = continuation;
160147
setCallback(next);
161-
setCancelCallback(continuation.getCancelCallback());
162148
start();
163149
}
164150

‎AndroidAsync/src/com/koushikdutta/async/future/Future.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
package com.koushikdutta.async.future;
2+
3+
4+
public interface Future<T> extends Cancellable, java.util.concurrent.Future<T> {
5+
6+
}

‎AndroidAsync/src/com/koushikdutta/async/future/SimpleCancelable.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
package com.koushikdutta.async.future;
2+
3+
public class SimpleCancelable implements Cancellable {
4+
boolean complete;
5+
@Override
6+
public boolean isDone() {
7+
return complete;
8+
}
9+
10+
public boolean setComplete() {
11+
synchronized (this) {
12+
if (canceled)
13+
return false;
14+
complete = true;
15+
}
16+
return true;
17+
}
18+
19+
@Override
20+
public boolean cancel() {
21+
synchronized (this) {
22+
if (complete)
23+
return false;
24+
if (canceled)
25+
return true;
26+
canceled = true;
27+
}
28+
if (parent != null)
29+
parent.cancel();
30+
return true;
31+
}
32+
boolean canceled;
33+
34+
Cancellable parent;
35+
public Cancellable getParent() {
36+
return parent;
37+
}
38+
39+
public void setParent(Cancellable parent) {
40+
this.parent = parent;
41+
}
42+
43+
@Override
44+
public boolean isCancelled() {
45+
return canceled || (parent != null && parent.isCancelled());
46+
}
47+
48+
public static final Cancellable COMPLETED = new SimpleCancelable() {
49+
{
50+
setComplete();
51+
}
52+
};
53+
}

‎AndroidAsync/src/com/koushikdutta/async/future/SimpleFuture.java‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
package com.koushikdutta.async.future;
2+
3+
import java.util.concurrent.ExecutionException;
4+
import java.util.concurrent.TimeUnit;
5+
import java.util.concurrent.TimeoutException;
6+
7+
import com.koushikdutta.async.AsyncServer.AsyncSemaphore;
8+
9+
public class SimpleFuture<T> extends SimpleCancelable implements Future<T> {
10+
@Override
11+
public boolean cancel(boolean mayInterruptIfRunning) {
12+
return cancel();
13+
}
14+
15+
AsyncSemaphore waiter;
16+
@Override
17+
public T get() throws InterruptedException, ExecutionException {
18+
synchronized (this) {
19+
if (isCancelled())
20+
return null;
21+
if (isDone())
22+
return getResult();
23+
if (waiter == null)
24+
waiter = new AsyncSemaphore();
25+
}
26+
waiter.acquire();
27+
return getResult();
28+
}
29+
30+
private T getResult() throws ExecutionException {
31+
if (exception != null)
32+
throw new ExecutionException(exception);
33+
return result;
34+
}
35+
36+
@Override
37+
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
38+
synchronized (this) {
39+
if (isCancelled())
40+
return null;
41+
if (isDone())
42+
return getResult();
43+
if (waiter == null)
44+
waiter = new AsyncSemaphore();
45+
}
46+
if (!waiter.tryAcquire(timeout, unit))
47+
return null;
48+
return getResult();
49+
}
50+
51+
@Override
52+
public boolean setComplete() {
53+
return setComplete((T)null);
54+
}
55+
56+
57+
Exception exception;
58+
public boolean setComplete(Exception e) {
59+
synchronized (this) {
60+
if (!super.setComplete())
61+
return false;
62+
if (waiter != null)
63+
waiter.release();
64+
exception = e;
65+
return true;
66+
}
67+
}
68+
69+
T result;
70+
public boolean setComplete(T value) {
71+
synchronized (this) {
72+
if (!super.setComplete())
73+
return false;
74+
result = value;
75+
if (waiter != null)
76+
waiter.release();
77+
return true;
78+
}
79+
}
80+
}

0 commit comments

Comments
 (0)