super(type, link, data);
}
@Override
protected void doTask(final Task task) throws Exception {
TransactionLegacy txn = TransactionLegacy.open(TransactionLegacy.CLOUD_DB);
try {
if (task.getType() != Task.Type.DATA) {
super.doTask(task);
return;
}
final byte[] data = task.getData();
Version ver = Request.getVersion(data);
if (ver.ordinal() != Version.v1.ordinal() && ver.ordinal() != Version.v3.ordinal()) {
s_logger.warn("Wrong version for clustered agent request");
super.doTask(task);
return;
}
long hostId = Request.getAgentId(data);
Link link = task.getLink();
if (Request.fromServer(data)) {
AgentAttache agent = findAttache(hostId);
if (Request.isControl(data)) {
if (agent == null) {
logD(data, "No attache to process cancellation");
return;
}
Request req = Request.parse(data);
Command[] cmds = req.getCommands();
CancelCommand cancel = (CancelCommand)cmds[0];
if (s_logger.isDebugEnabled()) {
logD(data, "Cancel request received");
}
agent.cancel(cancel.getSequence());
final Long current = agent._currentSequence;
// if the request is the current request, always have to trigger sending next request in
// sequence,
// otherwise the agent queue will be blocked
if (req.executeInSequence() && (current != null && current == Request.getSequence(data))) {
agent.sendNext(Request.getSequence(data));
}
return;
}
try {
if (agent == null || agent.isClosed()) {
throw new AgentUnavailableException("Unable to route to agent ", hostId);
}
if (Request.isRequest(data) && Request.requiresSequentialExecution(data)) {
// route it to the agent.
// But we have the serialize the control commands here so we have
// to deserialize this and send it through the agent attache.
Request req = Request.parse(data);
agent.send(req, null);
return;
} else {
if (agent instanceof Routable) {
Routable cluster = (Routable)agent;
cluster.routeToAgent(data);
} else {
agent.send(Request.parse(data));
}
return;
}
} catch (AgentUnavailableException e) {
logD(data, e.getMessage());
cancel(Long.toString(Request.getManagementServerId(data)), hostId, Request.getSequence(data), e.getMessage());
}
} else {
long mgmtId = Request.getManagementServerId(data);
if (mgmtId != -1 && mgmtId != _nodeId) {
routeToPeer(Long.toString(mgmtId), data);
if (Request.requiresSequentialExecution(data)) {
AgentAttache attache = (AgentAttache)link.attachment();
if (attache != null) {
attache.sendNext(Request.getSequence(data));
} else if (s_logger.isDebugEnabled()) {
logD(data, "No attache to process " + Request.parse(data).toString());
}
}
return;
} else {
if (Request.isRequest(data)) {
super.doTask(task);
} else {
// received an answer.
final Response response = Response.parse(data);
AgentAttache attache = findAttache(response.getAgentId());
if (attache == null) {
s_logger.info("SeqA " + response.getAgentId() + "-" + response.getSequence() + "Unable to find attache to forward " + response.toString());
return;
}
if (!attache.processAnswers(response.getSequence(), response)) {
s_logger.info("SeqA " + attache.getId() + "-" + response.getSequence() + ": Response is not processed: " + response.toString());
}
}
return;
}
}
} finally {
txn.close();
}
}