Method operator()

cpp/src/arrow/c/bridge.cc:2742–2769

  };

  Status operator()(const std::shared_ptr<RecordBatch>& record) {
    std::unique_lock<std::mutex> lock(state_->mutex_);
    if (state_->pending_requests_ == 0) {
      state_->cv_.wait(lock, [this]() -> bool {
        return !state_->error_.ok() || state_->pending_requests_ > 0;
      });
    }

    if (!state_->error_.ok()) {
      return state_->error_;
    }

    if (state_->pending_requests_ > 0) {
      state_->pending_requests_--;
      lock.unlock();

      ArrowAsyncTask task;
      task.private_data = new PrivateTaskData{state_, record};
      task.extract_data = AsyncProducer::extract_data;

      if (int status = handler_->on_next_task(handler_, &task, nullptr) != 0) {
        delete reinterpret_cast<PrivateTaskData*>(task.private_data);
        return Status::UnknownError("Received error from handler::on_next_task ", status);
      }
    }

    return Status::OK();
  }

  static void request(struct ArrowAsyncProducer* producer, int64_t n) {
    auto* self = reinterpret_cast<State*>(producer->private_data);