void ClientManagerImpl::recallMessage()

in cpp/source/client/ClientManagerImpl.cpp [1306:1405]
85 lines of code
14 McCabe index (conditional complexity)

void ClientManagerImpl::recallMessage(const std::string& target_host, const Metadata& metadata,
                                      const RecallMessageRequest& request, std::chrono::milliseconds timeout,
                                      const std::function<void(const std::error_code&, const RecallMessageResponse&)>& cb) {

  SPDLOG_DEBUG("RecallMessage Request: {}", request.ShortDebugString());

  RpcClientSharedPtr client = getRpcClient(target_host);
  if (!client) {
    SPDLOG_WARN("No RPC client for {}", target_host);
    RecallMessageResponse response;
    std::error_code ec = ErrorCode::BadRequest;
    cb(ec, response);
    return;
  }

  auto invocation_context = new InvocationContext<RecallMessageResponse>();
  invocation_context->task_name = fmt::format("Recall message, topic={}, recall handle={} to {}",
                                              request.topic().ShortDebugString(), request.recall_handle().data(), target_host);
  invocation_context->remote_address = target_host;
  for (const auto& item : metadata) {
    invocation_context->context.AddMetadata(item.first, item.second);
  }
  invocation_context->context.set_deadline(std::chrono::system_clock::now() + timeout);

  auto callback =
      [target_host, cb](const InvocationContext<RecallMessageResponse>* invocation_context) {

    std::error_code ec;
    if (!invocation_context->status.ok()) {
      SPDLOG_WARN("Failed to write EndTransaction to wire. gRPC-code: {}, gRPC-message: {}, host={}",
                  invocation_context->status.error_code(), invocation_context->status.error_message(),
                  invocation_context->remote_address);
      ec = ErrorCode::BadRequest;
      cb(ec, invocation_context->response);
      return;
    }

    auto&& status = invocation_context->response.status();
    auto&& peer_address = invocation_context->remote_address;
    switch (status.code()) {
      case rmq::Code::OK: {
        SPDLOG_DEBUG("Recall message OK. Response: {}, host={}",
                     invocation_context->response.ShortDebugString(), peer_address);
        break;
      }

      case rmq::Code::ILLEGAL_TOPIC: {
        SPDLOG_WARN("IllegalTopic: {}, host={}", status.message(), peer_address);
        ec = ErrorCode::IllegalTopic;
        break;
      }

      case rmq::Code::CLIENT_ID_REQUIRED: {
        SPDLOG_WARN("ClientIdRequired: {}, host={}", status.message(), peer_address);
        ec = ErrorCode::InternalClientError;
        break;
      }

      case rmq::Code::TOPIC_NOT_FOUND: {
        SPDLOG_WARN("TopicNotFound: {}, host={}", status.message(), peer_address);
        ec = ErrorCode::TopicNotFound;
        break;
      }

      case rmq::Code::UNAUTHORIZED: {
        SPDLOG_WARN("Unauthorized: {}, host={}", status.message(), peer_address);
        ec = ErrorCode::Unauthorized;
        break;
      }

      case rmq::Code::FORBIDDEN: {
        SPDLOG_WARN("Forbidden: {}, host={}", status.message(), peer_address);
        ec = ErrorCode::Forbidden;
        break;
      }

      case rmq::Code::INTERNAL_SERVER_ERROR: {
        SPDLOG_WARN("InternalServerError: {}, host={}", status.message(), peer_address);
        ec = ErrorCode::InternalServerError;
        break;
      }

      case rmq::Code::PROXY_TIMEOUT: {
        SPDLOG_WARN("GatewayTimeout: {}, host={}", status.message(), peer_address);
        ec = ErrorCode::GatewayTimeout;
        break;
      }

      default: {
        SPDLOG_WARN("NotSupported: please upgrade SDK to latest release. {}, host={}", status.message(), peer_address);
        ec = ErrorCode::NotSupported;
        break;
      }
    }
    cb(ec, invocation_context->response);
  };

  invocation_context->callback = callback;
  client->asyncRecallMessage(request, invocation_context);
}