Method executePipeline

src/brpc/couchbase.cpp:2489–2624

  return true;
}
std::vector<CouchbaseOperations::Result> CouchbaseOperations::executePipeline() {
  std::vector<CouchbaseOperations::Result> results;

  if (!pipeline_active || pipeline_operations_queue.empty()) {
    DEBUG_PRINT("No pipeline active or no operations queued");
    return results;
  }

  brpc::Controller cntl;
  channel_->CallMethod(NULL, &cntl, &pipeline_request_couchbase_req,
                       &pipeline_response_couchbase_resp, NULL);

  if (cntl.Failed()) {
    DEBUG_PRINT("Pipeline execution failed: " << cntl.ErrorText());
    // Create failure results for all operations
    size_t op_count = pipeline_operations_queue.size();
    results.reserve(op_count);

    CouchbaseOperations::Result failure_result;
    failure_result.success = false;
    failure_result.error_message = cntl.ErrorText();

    for (size_t i = 0; i < op_count; ++i) {
      results.push_back(failure_result);
    }

    clearPipeline();
    return results;
  }

  // Process each operation in the order they were added
  CouchbaseOperations::CouchbaseResponse* response =
      &pipeline_response_couchbase_resp;
  while (!pipeline_operations_queue.empty()) {
    CouchbaseOperations::Result result;
    operation_type op_type = pipeline_operations_queue.front();
    pipeline_operations_queue.pop();
    switch (op_type) {
      case GET: {
        std::string value;
        uint32_t flags = 0;
        uint64_t cas = 0;
        if (response->popGet(&value, &flags, &cas) == false) {
          result.success = false;
          result.value = "";
          result.error_message = response->lastError();
          result.status_code = response->_status_code;
        } else {
          result.success = true;
          result.value = value;
        }
        results.push_back(result);
        break;
      }
      case UPSERT: {
        if (response->popUpsert(NULL) == false) {
          result.success = false;
          result.value = "";