Method AsyncRequest

be/src/kudu/rpc/proxy.cc:192–249

}

void Proxy::AsyncRequest(const string& method,
                         const google::protobuf::Message& req,
                         google::protobuf::Message* response,
                         RpcController* controller,
                         const ResponseCallback& callback) {
  CHECK(!controller->call_) << "Controller should be reset";
  base::subtle::NoBarrier_Store(&is_started_, true);
  // TODO(awong): it would be great if we didn't have to heap allocate the
  // payload.
  auto req_payload = RequestPayload::CreateRequestPayload(
      RemoteMethod{service_name_, method},
      req, controller->ReleaseOutboundSidecars());
  if (!dns_resolver_) {
    // NOTE: we don't expect the user-provided callback to free sidecars, so
    // make sure the outbound call frees it for us.
    EnqueueRequest(method, std::move(req_payload), response, controller, callback,
                   OutboundCall::CallbackBehavior::kFreeSidecars);
    return;
  }

  // If we haven't successfully initialized the remote, e.g. because the DNS
  // lookup failed, refresh the DNS entry and enqueue the request.
  bool remote_initialized;
  {
    std::lock_guard<simple_spinlock> l(lock_);
    remote_initialized = conn_id_.remote().is_initialized();
  }
  if (!remote_initialized) {
    RefreshDnsAndEnqueueRequest(method, std::move(req_payload), response, controller, callback);
    return;
  }

  // Otherwise, just enqueue the request, but retry if there's an error, since
  // it's possible the physical address of the host was changed. We only retry
  // once more before calling the callback.
  auto refresh_dns_and_cb = [this, &method, callback, response, controller] () {
    // TODO(awong): we should be more specific here -- consider having the RPC
    // layer set a flag in the controller that warrants a retry.
    if (PREDICT_FALSE(!controller->status().ok())) {
      KLOG_EVERY_N_SECS(WARNING, 5)
          << Substitute("Call had error, refreshing address and retrying: $0",
                        controller->status().ToString());
      auto req_payload = controller->ReleaseRequestPayload();
      controller->Reset();
      RefreshDnsAndEnqueueRequest(method, std::move(req_payload), response, controller, callback);
      return;
    }
    // For any other status, OK or otherwise, just run the callback.
    controller->FreeOutboundSidecars();
    SCOPED_WATCH_STACK(100);
    callback();
  };
  // Since we may end up using the request payload in the event of a retry,
  // ensure the outbound call doesn't free the sidecars, and instead free
  // manually from within our callback.
  EnqueueRequest(method, std::move(req_payload), response, controller, refresh_dns_and_cb,
                 OutboundCall::CallbackBehavior::kDontFreeSidecars);
}