Function ExportAsyncRecordBatchReader

cpp/src/arrow/c/bridge.cc:2829–2865

}

Future<> ExportAsyncRecordBatchReader(
    std::shared_ptr<Schema> schema,
    AsyncGenerator<std::shared_ptr<RecordBatch>> generator,
    DeviceAllocationType device_type, struct ArrowAsyncDeviceStreamHandler* handler) {
  if (!schema) {
    handler->on_error(handler, EINVAL, "Schema is null", nullptr);
    handler->release(handler);
    return Future<>::MakeFinished(Status::Invalid("Schema is null"));
  }

  struct ArrowSchema c_schema;
  SchemaExportGuard guard(&c_schema);

  auto status = ExportSchema(*schema, &c_schema);
  if (!status.ok()) {
    handler->on_error(handler, EINVAL, status.message().c_str(), nullptr);
    handler->release(handler);
    return Future<>::MakeFinished(status);
  }

  return VisitAsyncGenerator(generator, AsyncProducer{device_type, &c_schema, handler})
      .Then(
          [handler]() -> Status {
            int status = handler->on_next_task(handler, nullptr, nullptr);
            handler->release(handler);
            if (status != 0) {
              return Status::UnknownError("Received error from handler::on_next_task ",
                                          status);
            }
            return Status::OK();
          },
          [handler](const Status status) -> Status {
            handler->on_error(handler, EINVAL, status.message().c_str(), nullptr);
            handler->release(handler);
            return status;
          });
}

}  // namespace arrow