Method operator()

cpp/src/arrow/util/async_generator.h:1276–1379

        : state(std::move(state)), index(index), recursive(recursive) {}

    void operator()(const Result<T>& maybe_next_ref) {
      // An item has been delivered by one of the inner subscriptions
      Future<T> next_fut;
      const Result<T>* maybe_next = &maybe_next_ref;

      // When an item is delivered (and the caller has asked for it) we grab the
      // next item from the inner subscription.  To avoid this behavior leading to an
      // infinite loop (this can happen if the caller's callback asks for the next item)
      // we use a while loop.
      while (true) {
        Future<T> sink;
        bool sub_finished = maybe_next->ok() && IsIterationEnd(**maybe_next);
        bool pull_next_sub = false;
        bool was_broken = false;
        bool should_mark_gen_complete = false;
        bool should_mark_final_error = false;
        {
          auto guard = state->mutex.Lock();
          if (state->broken) {
            // We've errored out previously so ignore the result.  If anyone was waiting
            // for this they will get IterationEnd when we purge
            was_broken = true;
          } else {
            if (!sub_finished) {
              // There is a result to deliver.  Either we can deliver it now or we will
              // queue it up
              if (state->waiting_jobs.empty()) {
                state->delivered_jobs.push_back(std::make_shared<DeliveredJob>(
                    state->active_subscriptions[index], *maybe_next, index));
              } else {
                sink = std::move(*state->waiting_jobs.front());
                state->waiting_jobs.pop_front();
              }
            }

            // If this is the first error then we transition the state to a broken state
            if (!maybe_next->ok()) {
              should_mark_final_error = true;
              state->SignalErrorUnlocked(guard);
            }
          }

          // If we finished this inner subscription then we need to grab a new inner
          // subscription to take its spot.  If we can't (because we're broken or
          // exhausted) then we aren't going to be starting any new futures and so
          // the number of running subscriptions drops.
          pull_next_sub = sub_finished && !state->source_exhausted && !was_broken;
          if (sub_finished && !pull_next_sub) {
            state->num_running_subscriptions--;
          }
          // There are three situations we won't pull again.  If an error occurred or we
          // are already finished or if no one was waiting for our result and so we queued
          // it up.  We will decrement outstanding_requests and possibly mark the
          // generator completed.
          if (state->broken || (!sink.is_valid() && !sub_finished) ||
              (sub_finished && state->source_exhausted)) {
            if (state->MarkTaskFinishedUnlocked(guard)) {
              should_mark_gen_complete = true;