| 3095 | |
| 3096 | private: |
| 3097 | Status ProduceCallback(size_t thread_index) { |
| 3098 | if (done_.load()) return Status::OK(); |
| 3099 | ARROW_ASSIGN_OR_RAISE(std::optional<ExecBatch> maybe_batch, |
| 3100 | gen_->NextOrdersBatch(thread_index)); |
| 3101 | if (!maybe_batch.has_value()) { |
| 3102 | int64_t batches_generated = gen_->orders_batches_generated(); |
| 3103 | if (batches_generated == batches_outputted_.load()) { |
| 3104 | bool expected = false; |
| 3105 | if (done_.compare_exchange_strong(expected, true)) |
| 3106 | ARROW_RETURN_NOT_OK(finished_callback_(batches_outputted_.load())); |
| 3107 | return Status::OK(); |
| 3108 | } |
| 3109 | return schedule_callback_( |
| 3110 | [this](size_t thread_index) { return this->ProduceCallback(thread_index); }); |
| 3111 | } |
| 3112 | ExecBatch batch = std::move(*maybe_batch); |
| 3113 | ARROW_RETURN_NOT_OK(output_callback_(std::move(batch))); |
| 3114 | batches_outputted_++; |
| 3115 | return schedule_callback_( |
| 3116 | [this](size_t thread_index) { return this->ProduceCallback(thread_index); }); |
| 3117 | } |
| 3118 | |
| 3119 | OutputBatchCallback output_callback_; |
| 3120 | FinishedCallback finished_callback_; |
no test coverage detected