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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
|
package com.orbekk.same;
import static com.orbekk.same.StackTraceUtil.throwableToString;
import java.util.ArrayList;
import java.util.List;
import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.orbekk.same.State.Component;
import com.orbekk.util.DelayedOperation;
import com.orbekk.util.WorkQueue;
public class Client implements DiscoveryListener {
private Logger logger = LoggerFactory.getLogger(getClass());
private ConnectionManager connections;
private State state;
private String myUrl;
private List<StateChangedListener> stateListeners =
new ArrayList<StateChangedListener>();
private NetworkNotificationListener networkListener;
public class ClientInterfaceImpl implements ClientInterface {
private ClientInterfaceImpl() {
}
/** Get a copy of all the client state.
*/
@Override
public State getState() {
return new State(state);
}
// TODO: Do this asynchronously? Currently this is already achieved
// on Android, which makes the Java and Android versions different.
@Override
public DelayedOperation set(Component component) {
DelayedOperation op = new DelayedOperation();
String masterUrl = state.getDataOf(".masterUrl");
MasterService master = connections.getMaster(masterUrl);
try {
boolean success = master.updateStateRequest(
component.getName(), component.getData(),
component.getRevision());
if (success) {
op.complete(DelayedOperation.Status.createOk());
} else {
op.complete(DelayedOperation.Status
.createConflict("Conflict from master"));
}
} catch (Exception e) {
logger.error("Unable to contact master. Update fails.", e);
String e_ = throwableToString(e);
op.complete(DelayedOperation.Status.createError(
"Error contacting master. Update fails: " + e_));
}
return op;
}
@Override
public void addStateListener(StateChangedListener listener) {
stateListeners.add(listener);
}
@Override
public void removeStateListener(StateChangedListener listener) {
stateListeners.remove(listener);
}
}
private ClientInterface clientInterface = new ClientInterfaceImpl();
private ClientService serviceImpl = new ClientService() {
@Override
public void setState(String component, String data, long revision) throws Exception {
boolean status = state.update(component, data, revision);
if (status) {
for (StateChangedListener listener : stateListeners) {
listener.stateChanged(state.getComponent(component));
}
} else {
logger.warn("Ignoring update: {}",
new State.Component(component, revision, data));
}
}
@Override
public void notifyNetwork(String networkName, String masterUrl) throws Exception {
logger.info("NotifyNetwork(networkName={}, masterUrl={})",
networkName, masterUrl);
if (networkListener != null) {
networkListener.notifyNetwork(networkName, masterUrl);
}
}
@Override
public void discoveryRequest(String remoteUrl) {
discoveryThread.add(remoteUrl);
}
};
private WorkQueue<String> discoveryThread = new WorkQueue<String>() {
@Override protected void onChange() {
List<String> pending = getAndClear();
for (String url : pending) {
discover(url);
}
}
};
public Client(State state, ConnectionManager connections,
String myUrl) {
this.state = state;
this.connections = connections;
this.myUrl = myUrl;
}
public void start() {
discoveryThread.start();
}
public void interrupt() {
discoveryThread.interrupt();
}
public String getUrl() {
return myUrl;
}
public void joinNetwork(String masterUrl) {
logger.info("joinNetwork({})", masterUrl);
MasterService master = connections.getMaster(masterUrl);
state.clear();
try {
master.joinNetworkRequest(myUrl);
} catch (Exception e) {
logger.error("Unable to connect to master.", e);
}
}
ClientInterface getInterface() {
return clientInterface;
}
String lib_get(String name) {
return state.getDataOf(name);
}
<T> T lib_get(String name, TypeReference<T> type) {
return state.getParsedData(name, type);
}
void lib_set(String name, String data) throws UpdateConflict {
String masterUrl = state.getDataOf(".masterUrl");
long revision = state.getRevision(name) + 1;
MasterService master = connections.getMaster(masterUrl);
try {
boolean success = master.updateStateRequest(name, data,
revision);
if (!success) {
throw new UpdateConflict("State update conflict when " +
"updating " + name);
}
} catch (Exception e) {
logger.error("Unable to contact master. Update fails.", e);
throw new UpdateConflict("Unable to contact master. Update fails.");
}
}
public State.Component getState(String name) {
return state.getComponent(name);
}
State testGetState() {
return state;
}
public void setNetworkListener(NetworkNotificationListener listener) {
this.networkListener = listener;
}
public void sendDiscoveryRequest(String url) {
try {
connections.getClient(url)
.discoveryRequest(myUrl);
} catch (Exception e) {
logger.warn("Failed to send discovery request: {}",
throwableToString(e));
}
}
@Override
public void discover(String url) {
String networkName = state.getDataOf(".networkName");
if (networkName.equals(".InvalidClientNetwork")) {
logger.warn("Client not joined to a network. Ignoring discovery");
return;
} else if (networkName.equals(".Private")) {
logger.info("Ignoring broadcast to .Private network.");
return;
}
if (!url.equals(myUrl)) {
try {
connections.getClient(url)
.notifyNetwork(state.getDataOf(".networkName"),
state.getDataOf(".masterUrl"));
} catch (Exception e) {
logger.warn("Failed to contact new client {}: {}", url,
throwableToString(e));
}
}
}
public ClientService getService() {
return serviceImpl;
}
}
|
