Method next

cpp/src/arrow/c/bridge.cc:2537–2565

        : queue_size_{queue_size}, mapper_{std::move(mapper)} {}

    Result<RecordBatchWithMetadata> next() {
      TaskWithMetadata task;
      {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock,
                 [&] { return !error_.ok() || !batches_.empty() || end_of_stream_; });
        if (!error_.ok()) {
          return error_;
        }

        if (batches_.empty() && end_of_stream_) {
          return IterationEnd<RecordBatchWithMetadata>();
        }

        task = std::move(batches_.front());
        batches_.pop();
      }

      producer_->request(producer_, 1);
      ArrowDeviceArray out;
      if (task.task_.extract_data(&task.task_, &out) != 0) {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock, [&] { return !error_.ok(); });
        return error_;
      }

      ARROW_ASSIGN_OR_RAISE(auto batch, ImportDeviceRecordBatch(&out, schema_, mapper_));
      return RecordBatchWithMetadata{std::move(batch), std::move(task.metadata_)};
    }

    const uint64_t queue_size_;
    const DeviceMemoryMapper mapper_;