Skip to content

Commit d8fa4bb

Browse files
koushkoush
committed
wip
1 parent 3638a69 commit d8fa4bb

9 files changed

Lines changed: 260 additions & 54 deletions

File tree

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

Lines changed: 20 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -123,21 +123,26 @@ public void run() {
123123
}
124124
}
125125

126-
private static void runQueue(LinkedList<Scheduled> queue) {
126+
private static final long DEFAULT_WAIT = 100;
127+
private static long runQueue(LinkedList<Scheduled> queue) {
128+
long wait = DEFAULT_WAIT;
127129
long now = System.currentTimeMillis();
128130
LinkedList<Scheduled> later = null;
129131
while (queue.size() > 0) {
130132
Scheduled s = queue.remove();
131133
if (s.time < now)
132134
s.runnable.run();
133135
else {
136+
wait = Math.min(wait, s.time - now);
134137
if (later == null)
135138
later = new LinkedList<AsyncServer.Scheduled>();
136139
later.add(s);
137140
}
138141
}
139142
if (later != null)
140143
queue.addAll(later);
144+
145+
return wait;
141146
}
142147

143148
private static class Scheduled {
@@ -235,15 +240,17 @@ private SimpleCancelable prepareConnectSocketCancelable(final SocketChannel sock
235240
return new SimpleCancelable() {
236241
@Override
237242
public Cancelable cancel() {
238-
synchronized (this) {
239-
super.cancel();
240-
try {
241-
sc.close();
242-
}
243-
catch (IOException e) {
243+
run(new Runnable() {
244+
@Override
245+
public void run() {
246+
try {
247+
sc.close();
248+
}
249+
catch (IOException e) {
250+
}
244251
}
245-
return this;
246-
}
252+
});
253+
return this;
247254
}
248255
};
249256
}
@@ -410,21 +417,21 @@ private static void shutdownEverything(Selector selector) {
410417
}
411418
}
412419

413-
private static void lockAndRunQueue(AsyncServer server, LinkedList<Scheduled> queue) {
420+
private static long lockAndRunQueue(AsyncServer server, LinkedList<Scheduled> queue) {
414421
LinkedList<Scheduled> copy;
415422
synchronized (server) {
416423
copy = new LinkedList<Scheduled>(queue);
417424
queue.clear();
418425
}
419-
runQueue(copy);
426+
return runQueue(copy);
420427
}
421428

422429
private static void runLoop(AsyncServer server, Selector selector, LinkedList<Scheduled> queue, boolean keepRunning) throws IOException {
423430
// Log.i(LOGTAG, "Keys: " + selector.keys().size());
424431
boolean needsSelect = true;
425432

426433
// run the queue to populate the selector with keys
427-
lockAndRunQueue(server, queue);
434+
long wait = lockAndRunQueue(server, queue);
428435
synchronized (server) {
429436
// select now to see if anything is ready immediately. this
430437
// also clears the canceled key queue.
@@ -444,7 +451,7 @@ private static void runLoop(AsyncServer server, Selector selector, LinkedList<Sc
444451

445452
if (needsSelect) {
446453
// nothing to select immediately but there so let's block and wait.
447-
selector.select(100);
454+
selector.select(wait);
448455
}
449456

450457
// process whatever keys are ready

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

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,11 +8,27 @@ public class BufferedDataEmitter implements DataEmitter, DataCallback {
88
public BufferedDataEmitter(DataEmitter emitter) {
99
mEmitter = emitter;
1010
mEmitter.setDataCallback(this);
11+
12+
mEmitter.setEndCallback(new CompletedCallback() {
13+
@Override
14+
public void onCompleted(Exception ex) {
15+
mEnded = true;
16+
mEndException = ex;
17+
if (mBuffers.remaining() == 0 && mEndCallback != null)
18+
mEndCallback.onCompleted(ex);
19+
}
20+
});
1121
}
1222

23+
boolean mEnded = false;
24+
Exception mEndException;
25+
1326
public void onDataAvailable() {
1427
if (mDataCallback != null && !mPaused && mBuffers.remaining() > 0)
1528
mDataCallback.onDataAvailable(this, mBuffers);
29+
30+
if (mEnded && mBuffers.remaining() == 0)
31+
mEndCallback.onCompleted(mEndException);
1632
}
1733

1834
ByteBufferList mBuffers = new ByteBufferList();
@@ -60,13 +76,15 @@ public boolean isPaused() {
6076
return mPaused;
6177
}
6278

79+
80+
CompletedCallback mEndCallback;
6381
@Override
6482
public void setEndCallback(CompletedCallback callback) {
65-
mEmitter.setEndCallback(callback);
83+
mEndCallback = callback;
6684
}
6785

6886
@Override
6987
public CompletedCallback getEndCallback() {
70-
return mEmitter.getEndCallback();
88+
return mEndCallback;
7189
}
7290
}

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

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
package com.koushikdutta.async;
2+
3+
import com.koushikdutta.async.callback.CompletedCallback;
4+
import com.koushikdutta.async.callback.DataCallback;
5+
6+
public class FilteredDataEmitter implements DataEmitter {
7+
DataEmitter mEmitter;
8+
public DataEmitter getDataEmitter() {
9+
return mEmitter;
10+
}
11+
12+
public void setDataEmitter(DataEmitter emitter) {
13+
if (mEmitter != null) {
14+
mEmitter.setDataCallback(null);
15+
}
16+
mEmitter = emitter;
17+
mEmitter.setDataCallback(new DataCallback() {
18+
@Override
19+
public void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
20+
FilteredDataEmitter.this.onDataAvailable(emitter, bb);
21+
}
22+
});
23+
}
24+
25+
protected void onDataAvailable(DataEmitter emitter, ByteBufferList bb) {
26+
Util.emitAllData(this, bb);
27+
// if there's data after the emitting, and it is paused... the underlying implementation
28+
// is obligated to cache the byte buffer list.
29+
}
30+
31+
DataCallback mDataCallback;
32+
@Override
33+
public void setDataCallback(DataCallback callback) {
34+
mDataCallback = callback;
35+
}
36+
37+
@Override
38+
public DataCallback getDataCallback() {
39+
return mDataCallback;
40+
}
41+
42+
@Override
43+
public boolean isChunked() {
44+
return mEmitter.isChunked();
45+
}
46+
47+
@Override
48+
public void pause() {
49+
mEmitter.pause();
50+
}
51+
52+
@Override
53+
public void resume() {
54+
mEmitter.resume();
55+
}
56+
57+
@Override
58+
public boolean isPaused() {
59+
return mEmitter.isPaused();
60+
}
61+
62+
@Override
63+
public void setEndCallback(CompletedCallback callback) {
64+
mEmitter.setEndCallback(callback);
65+
}
66+
67+
@Override
68+
public CompletedCallback getEndCallback() {
69+
return mEmitter.getEndCallback();
70+
}
71+
}

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

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
package com.koushikdutta.async;
2+
3+
import java.nio.ByteBuffer;
4+
5+
import com.koushikdutta.async.callback.CompletedCallback;
6+
import com.koushikdutta.async.callback.DataCallback;
7+
import com.koushikdutta.async.callback.WritableCallback;
8+
9+
public class WrapperSocket implements AsyncSocket {
10+
private AsyncSocket mSocket;
11+
public void setSocket(AsyncSocket socket) {
12+
mSocket = socket;
13+
}
14+
15+
public AsyncSocket getSocket() {
16+
return mSocket;
17+
}
18+
19+
@Override
20+
public void setDataCallback(DataCallback callback) {
21+
mSocket.setDataCallback(callback);
22+
}
23+
24+
@Override
25+
public DataCallback getDataCallback() {
26+
return mSocket.getDataCallback();
27+
}
28+
29+
@Override
30+
public boolean isChunked() {
31+
return mSocket.isChunked();
32+
}
33+
34+
@Override
35+
public void pause() {
36+
mSocket.pause();
37+
}
38+
39+
@Override
40+
public void resume() {
41+
mSocket.resume();
42+
}
43+
44+
@Override
45+
public boolean isPaused() {
46+
return mSocket.isPaused();
47+
}
48+
49+
@Override
50+
public void setEndCallback(CompletedCallback callback) {
51+
mSocket.setEndCallback(callback);
52+
}
53+
54+
@Override
55+
public CompletedCallback getEndCallback() {
56+
return mSocket.getEndCallback();
57+
}
58+
59+
@Override
60+
public void write(ByteBuffer bb) {
61+
mSocket.write(bb);
62+
}
63+
64+
@Override
65+
public void write(ByteBufferList bb) {
66+
mSocket.write(bb);
67+
}
68+
69+
@Override
70+
public void setWriteableCallback(WritableCallback handler) {
71+
mSocket.setWriteableCallback(handler);
72+
}
73+
74+
@Override
75+
public WritableCallback getWriteableCallback() {
76+
return getWriteableCallback();
77+
}
78+
79+
@Override
80+
public boolean isOpen() {
81+
return mSocket.isOpen();
82+
}
83+
84+
@Override
85+
public void close() {
86+
mSocket.close();
87+
}
88+
89+
@Override
90+
public void setClosedCallback(CompletedCallback handler) {
91+
mSocket.setClosedCallback(handler);
92+
}
93+
94+
@Override
95+
public CompletedCallback getClosedCallback() {
96+
return mSocket.getClosedCallback();
97+
}
98+
99+
@Override
100+
public AsyncServer getServer() {
101+
return mSocket.getServer();
102+
}
103+
104+
}

0 commit comments

Comments
 (0)