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
|
package com.koushikdutta.async;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import junit.framework.Assert;
import com.koushikdutta.async.callback.CompletedCallback;
import com.koushikdutta.async.callback.DataCallback;
import com.koushikdutta.async.callback.WritableCallback;
public class Util {
public static void emitAllData(DataEmitter emitter, ByteBufferList list) {
int remaining;
AsyncSocket socket = null;
if (emitter instanceof AsyncSocket)
socket = (AsyncSocket)emitter;
DataCallback handler;
while ((socket == null || !socket.isPaused()) && (handler = emitter.getDataCallback()) != null && (remaining = list.remaining()) > 0) {
handler.onDataAvailable(emitter, list);
if (remaining == list.remaining() && handler == emitter.getDataCallback()) {
Assert.fail("mDataHandler failed to consume data, yet remains the mDataHandler.");
break;
}
}
Assert.assertTrue(list.remaining() == 0 || (socket != null && socket.isPaused()));
}
public static void emitAllData(DataEmitter emitter, ByteBuffer b) {
ByteBufferList list = new ByteBufferList();
list.add(b);
emitAllData(emitter, list);
// previous call makes sure list is empty,
// so this is safe to clear
b.position(b.limit());
}
public static void pump(final InputStream is, final DataSink ds, final CompletedCallback callback) {
final WritableCallback cb = new WritableCallback() {
private void close() {
try {
is.close();
}
catch (IOException e) {
e.printStackTrace();
}
}
byte[] buffer = new byte[8192];
ByteBuffer pending = ByteBuffer.wrap(buffer);
{
pending.limit(pending.position());
}
@Override
public void onWriteable() {
try {
int remaining;
do {
if (pending.remaining() == 0) {
int read = is.read(buffer);
if (read == -1) {
close();
callback.onCompleted(null);
return;
}
pending.position(0);
pending.limit(read);
}
remaining = pending.remaining();
ds.write(pending);
}
while (remaining != pending.remaining());
}
catch (Exception e) {
close();
callback.onCompleted(e);
return;
}
}
};
ds.setWriteableCallback(cb);
cb.onWriteable();
}
public static void writeAll(final DataSink sink, final ByteBufferList bb, final CompletedCallback callback) {
sink.setWriteableCallback(new WritableCallback() {
@Override
public void onWriteable() {
if (bb.remaining() == 0)
return;
sink.write(bb);
if (bb.remaining() == 0 && callback != null)
callback.onCompleted(null);
}
});
sink.write(bb);
if (bb.remaining() == 0 && callback != null)
callback.onCompleted(null);
}
public static void writeAll(DataSink sink, byte[] bytes, CompletedCallback callback) {
ByteBuffer bb = ByteBuffer.wrap(bytes);
ByteBufferList bbl = new ByteBufferList();
bbl.add(bb);
writeAll(sink, bbl, callback);
}
}
|
