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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
/*
* Copyright (c) 2010 Remko Tron¨on
* Licensed under the GNU General Public License v3.
* See Documentation/Licenses/GPLv3.txt for more information.
*/
/*
* Copyright (c) 2010-2011, Isode Limited, London, England.
* All rights reserved.
*/
package com.isode.stroke.network;
import com.isode.stroke.base.ByteArray;
import com.isode.stroke.eventloop.Event.Callback;
import com.isode.stroke.eventloop.EventLoop;
import com.isode.stroke.eventloop.EventOwner;
import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
public class JavaConnection extends Connection implements EventOwner {
private class Worker implements Runnable {
private final HostAddressPort address_;
private OutputStream write_;
private BufferedReader read_;
private final List<ByteArray> writeBuffer_ = Collections.synchronizedList(new ArrayList<ByteArray>());
public Worker(HostAddressPort address) {
address_ = address;
}
public void run() {
try {
socket_ = new Socket(address_.getAddress().getInetAddress(), address_.getPort());
write_ = socket_.getOutputStream();
read_ = new BufferedReader(new InputStreamReader(socket_.getInputStream(), "utf-8"));
} catch (IOException ex) {
handleConnected(true);
return;
}
handleConnected(false);
while (!disconnecting_) {
boolean didWrite = false;
while (!writeBuffer_.isEmpty()) {
didWrite = true;
ByteArray data = writeBuffer_.get(0);
for (byte datum : data.getData()) {
try {
write_.write(datum);
} catch (IOException ex) {
disconnecting_ = true;
handleDisconnected(Error.WriteError);
}
}
writeBuffer_.remove(0);
}
if (didWrite && !disconnecting_) {
try {
write_.flush();
} catch (IOException ex) {
disconnecting_ = true;
handleDisconnected(Error.WriteError);
}
}
ByteArray data = new ByteArray();
try {
while (read_.ready()) {
char[] c = new char[1024];
int i = read_.read(c, 0, c.length);
if (i > 0) {
data.append(new String(c, 0, i));
}
}
} catch (IOException ex) {
handleDisconnected(Error.ReadError);
return;
}
if (!data.isEmpty()) {
handleDataRead(data);
}
try {
Thread.sleep(100);
} catch (InterruptedException ex) {
/* We've been woken up, probably to force us to do something.*/
}
}
try {
read_.close();
write_.close();
socket_.close();
} catch (IOException ex) {
/* Do we need to return an error if we're already trying to close? */
}
}
private void handleConnected(final boolean error) {
eventLoop_.postEvent(new Callback() {
public void run() {
onConnectFinished.emit(error);
}
});
}
private void handleDisconnected(final Error error) {
eventLoop_.postEvent(new Callback() {
public void run() {
onDisconnected.emit(error);
}
});
}
private void handleDataRead(final ByteArray data) {
eventLoop_.postEvent(new Callback() {
public void run() {
onDataRead.emit(data);
}
});
}
public void write(ByteArray data) {
writeBuffer_.add(data);
}
}
private JavaConnection(EventLoop eventLoop) {
eventLoop_ = eventLoop;
}
public static JavaConnection create(EventLoop eventLoop) {
return new JavaConnection(eventLoop);
}
@Override
public void listen() {
//TODO: needed for server, not for client.
throw new UnsupportedOperationException("Not supported yet.");
}
@Override
public void connect(HostAddressPort address) {
worker_ = new Worker(address);
Thread workerThread = new Thread(worker_);
workerThread.setDaemon(true);
workerThread.start();
}
@Override
public void disconnect() {
disconnecting_ = true;
}
@Override
public void write(ByteArray data) {
worker_.writeBuffer_.add(data);
}
@Override
public HostAddressPort getLocalAddress() {
return new HostAddressPort(new HostAddress(socket_.getLocalAddress()), socket_.getLocalPort());
}
private final EventLoop eventLoop_;
private boolean disconnecting_ = false;
private Socket socket_;
private Worker worker_;
}
|
Swift