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
|
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.util.WorkQueue;
public class ClientServiceImpl implements UrlReceiver,
DiscoveryListener {
private Logger logger = LoggerFactory.getLogger(getClass());
private ConnectionManager connections;
private State state;
private String myUrl = null;
private StateChangedListener stateListener;
private NetworkNotificationListener networkListener;
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) {
if (stateListener != null) {
stateListener.stateChanged(component, data);
}
} 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 ClientServiceImpl(State state, ConnectionManager connections) {
this.state = state;
this.connections = connections;
}
public void start() {
discoveryThread.start();
}
public void interrupt() {
discoveryThread.interrupt();
}
@Override
public void setUrl(String myUrl) {
this.myUrl = myUrl + "ClientService.json";
logger.info("My URL is {}.", this.myUrl);
}
public String getUrl() {
return myUrl;
}
public void joinNetwork(String masterUrl) {
if (myUrl != null) {
MasterService master = connections.getMaster(masterUrl);
state.clear();
try {
master.joinNetworkRequest(myUrl);
} catch (Exception e) {
logger.error("Unable to connect to master.", e);
}
} else {
logger.error("Tried to join network at {}, but my url is unknown. " +
"Run discovery service.", masterUrl);
}
}
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 setStateChangedListener(StateChangedListener listener) {
this.stateListener = listener;
}
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) {
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;
}
}
|
