in cpp/source/client/ClientManagerImpl.cpp [562:666]
void ClientManagerImpl::resolveRoute(const std::string& target_host,
const Metadata& metadata,
const QueryRouteRequest& request,
std::chrono::milliseconds timeout,
const std::function<void(const std::error_code&, const TopicRouteDataPtr&)>& cb) {
SPDLOG_DEBUG("Name server connection URL: {}", target_host);
SPDLOG_DEBUG("Query route request: {}", request.ShortDebugString());
RpcClientSharedPtr client = getRpcClient(target_host, false);
if (!client) {
SPDLOG_WARN("Failed to create RPC client for name server[host={}]", target_host);
std::error_code ec = ErrorCode::RequestTimeout;
cb(ec, nullptr);
return;
}
auto invocation_context = new InvocationContext<QueryRouteResponse>();
invocation_context->task_name = fmt::format("Query route of topic={} from {}", request.topic().name(), target_host);
invocation_context->remote_address = target_host;
invocation_context->context.set_deadline(std::chrono::system_clock::now() + timeout);
for (const auto& item : metadata) {
invocation_context->context.AddMetadata(item.first, item.second);
}
auto callback = [cb](const InvocationContext<QueryRouteResponse>* invocation_context) {
if (!invocation_context->status.ok()) {
SPDLOG_WARN("Failed to send query route request to server[host={}]. Reason: {}",
invocation_context->remote_address, invocation_context->status.error_message());
std::error_code ec = ErrorCode::RequestTimeout;
cb(ec, nullptr);
return;
}
std::error_code ec;
auto&& status = invocation_context->response.status();
switch (status.code()) {
case rmq::Code::OK: {
std::vector<rmq::MessageQueue> message_queues;
for (const auto& item : invocation_context->response.message_queues()) {
message_queues.push_back(item);
}
auto ptr = std::make_shared<TopicRouteData>(std::move(message_queues));
cb(ec, ptr);
break;
}
case rmq::Code::ILLEGAL_ACCESS_POINT: {
SPDLOG_WARN("IllegalAccessPoint: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::IllegalAccessPoint;
cb(ec, nullptr);
break;
}
case rmq::Code::UNAUTHORIZED: {
SPDLOG_WARN("Unauthorized: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::Unauthorized;
cb(ec, nullptr);
break;
}
case rmq::Code::TOPIC_NOT_FOUND: {
SPDLOG_WARN("TopicNotFound: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::NotFound;
cb(ec, nullptr);
break;
}
case rmq::Code::TOO_MANY_REQUESTS: {
SPDLOG_WARN("TooManyRequest: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::TooManyRequests;
cb(ec, nullptr);
break;
}
case rmq::Code::CLIENT_ID_REQUIRED: {
SPDLOG_ERROR("ClientIdRequired: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::InternalClientError;
cb(ec, nullptr);
break;
}
case rmq::Code::INTERNAL_SERVER_ERROR: {
SPDLOG_WARN("InternalServerError: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::InternalServerError;
cb(ec, nullptr);
break;
}
case rmq::Code::PROXY_TIMEOUT: {
SPDLOG_WARN("GatewayTimeout: {}. Host={}", status.message(), invocation_context->remote_address);
ec = ErrorCode::GatewayTimeout;
cb(ec, nullptr);
break;
}
default: {
SPDLOG_WARN("NotImplement: Please upgrade to latest SDK release. Host={}", invocation_context->remote_address);
ec = ErrorCode::NotImplemented;
cb(ec, nullptr);
break;
}
}
};
invocation_context->callback = callback;
client->asyncQueryRoute(request, invocation_context);
}