}
}
// Response is an ack to a prior PubSubRequest so first retrieve the
// PubSub data for this txn.
PubSubData pubSubData = txn2PubSubData.remove(response.getTxnId());
// Validate that the PubSub data for this txn is stored. If not, just
// log an error message and return since we don't know how to handle
// this.
if (pubSubData == null) {
logger.error("PubSub Data was not found for PubSubResponse: {}", response);
return;
}
// Store the topic2Host mapping if this wasn't a server redirect. We'll
// assume that if the server was able to have an open Channel connection
// to the client, and responded with an ack message other than the
// NOT_RESPONSIBLE_FOR_TOPIC one, it is the correct topic master.
if (!response.getStatusCode().equals(StatusCode.NOT_RESPONSIBLE_FOR_TOPIC)) {
// Retrieve the server host that we've connected to and store the
// mapping from the topic to this host. For all other non-redirected
// server statuses, we consider that as a successful connection to the
// correct topic master.
InetSocketAddress host = NetUtils.getHostFromChannel(ctx.getChannel());
channelManager.storeTopic2HostMapping(pubSubData.topic, host);
}
// Depending on the operation type, call the appropriate handler.
logger.debug("Handling a {} response: {}, pubSubData: {}, host: {}.",
va(pubSubData.operationType, response, pubSubData, ctx.getChannel()));
AbstractResponseHandler respHandler = handlers.get(pubSubData.operationType);
if (null == respHandler) {
// The above are the only expected PubSubResponse messages received
// from the server for the various client side requests made.
logger.error("Response received from server is for an unhandled operation {}, txnId: {}.",
va(pubSubData.operationType, response.getTxnId()));
pubSubData.getCallback().operationFailed(pubSubData.context,
new UnexpectedConditionException("Can't find response handler for operation "
+ pubSubData.operationType));
return;
}
respHandler.handleResponse(response, pubSubData, ctx.getChannel());