| 3157 | |
| 3158 | private: |
| 3159 | Status ProduceCallback(size_t thread_index) { |
| 3160 | if (done_.load()) return Status::OK(); |
| 3161 | ARROW_ASSIGN_OR_RAISE(std::optional<ExecBatch> maybe_batch, |
| 3162 | gen_->NextLineItemBatch(thread_index)); |
| 3163 | if (!maybe_batch.has_value()) { |
| 3164 | int64_t batches_generated = gen_->lineitem_batches_generated(); |
| 3165 | if (batches_generated == batches_outputted_.load()) { |
| 3166 | bool expected = false; |
| 3167 | if (done_.compare_exchange_strong(expected, true)) |
| 3168 | ARROW_RETURN_NOT_OK(finished_callback_(batches_outputted_.load())); |
| 3169 | return Status::OK(); |
| 3170 | } |
| 3171 | // We may have generated but not outputted all of the batches. |
| 3172 | return schedule_callback_( |
| 3173 | [this](size_t thread_index) { return this->ProduceCallback(thread_index); }); |
| 3174 | } |
| 3175 | ExecBatch batch = std::move(*maybe_batch); |
| 3176 | ARROW_RETURN_NOT_OK(output_callback_(std::move(batch))); |
| 3177 | batches_outputted_++; |
| 3178 | return schedule_callback_( |
| 3179 | [this](size_t thread_index) { return this->ProduceCallback(thread_index); }); |
| 3180 | } |
| 3181 | |
| 3182 | OutputBatchCallback output_callback_; |
| 3183 | FinishedCallback finished_callback_; |
no test coverage detected