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
|
package derms.replica1;
import derms.Replica;
import derms.ReplicaManager;
import derms.Request;
import derms.Response;
import derms.replica2.DermsLogger;
import derms.util.ThreadPool;
import javax.xml.ws.Endpoint;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;
public class Replica1 implements Replica {
private boolean alive = false;
private final ExecutorService pool;
private final Logger log;
private final InetAddress localAddr;
// private final ResponderClient responderClient;
// private final CoordinatorClient coordinatorClient;
private final String responderClientID = "MTL";
private final String coordinatorClientID = "MTLC1111";
private final ReplicaManager replicaManager;
private DERMSServer server;
private final ConcurrentHashMap<String, Integer> portsMap;
private final Random r = new Random();
public Replica1(ReplicaManager replicaManager) {
this.replicaManager = replicaManager;
pool = Executors.newFixedThreadPool(5);
try {
localAddr = InetAddress.getLocalHost();
} catch (UnknownHostException e) {
throw new RuntimeException(e);
}
// responderClient = new ResponderClient(responderClientID);
// coordinatorClient = new CoordinatorClient(coordinatorClientID);
try {
this.log = DermsLogger.getLogger(getClass());
} catch (IOException e) {
throw new RuntimeException(e);
}
portsMap = new ConcurrentHashMap<>();
portsMap.put("MTL", r.nextInt(60000-8000) + 8000);
portsMap.put("QUE", r.nextInt(60000-8000) + 8000);
portsMap.put("SHE", r.nextInt(60000-8000) + 8000);
}
@Override
public boolean isAlive() {
return alive;
}
@Override
public void startProcess() {
try {
server = new DERMSServer("MTL", portsMap);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
new DERMSServer("SHE", portsMap);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
new DERMSServer("QUE", portsMap);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
// try {
// Endpoint.publish("http://localhost:8387/ws/derms", new DERMSServer("MTL"));
// Endpoint.publish("http://localhost:8081/ws/derms", new DERMSServer("QUE"));
// Endpoint.publish("http://localhost:8082/ws/derms", new DERMSServer("SHE"));
// } catch (InterruptedException e) {}
alive = true;
log.info(getClass().getSimpleName() + " started.");
log.config("Local address is "+localAddr.toString());
}
@Override
public void processRequest(Request request) {
log.info(request.toString());
String status = "";
boolean isSuccess = true;
try {
switch (request.getFunction()) {
case "addResource":
status = server.addResource(request.getResourceID(), request.getResourceType(), request.getDuration());
break;
case "removeResource":
status = server.removeResource(request.getResourceID(), request.getDuration());
break;
case "listResourceAvailability":
// status = String.join(",", responderClient.listResourceAvailability(request.getResourceType()));
status = String.join(",", server.listResourceAvailability(request.getResourceType()));
break;
case "requestResource":
status = server.requestResource(coordinatorClientID, request.getResourceID(), request.getDuration());
break;
case "findResource":
status = String.join(",", server.findResource(coordinatorClientID, request.getResourceType()));
break;
case "returnResource":
status = server.returnResource(coordinatorClientID, request.getResourceID());
break;
case "swapResource":
status = server.swapResource(coordinatorClientID, request.getOldResourceID(), request.getOldResourceType(), request.getResourceID(), request.getResourceType());
break;
default:
status = "Failure: unknown function '" + request.getFunction() + "'";
}
} catch (Exception e) {
log.warning(e.getMessage());
status = "Failure: " + request.getFunction() + ": " + e.getMessage();
isSuccess = false;
}
Response response = new Response(request, replicaManager.getReplicaId(), status, isSuccess); // TODO: isSuccess flag
log.info("Processed request " + request + "; response: " + response);
replicaManager.sendResponseToFE(response);
}
@Override
public void restart() {
log.info("Shutting down...");
ThreadPool.shutdown(pool, log);
alive = false;
log.info("Finished shutting down.");
startProcess();
}
@Override
public int getId() {
return 1;
}
}
|