| 1385 | |
| 1386 | struct OuterCallback { |
| 1387 | void operator()(const Result<AsyncGenerator<T>>& initial_maybe_next) { |
| 1388 | Result<AsyncGenerator<T>> maybe_next = initial_maybe_next; |
| 1389 | while (true) { |
| 1390 | // We have been given a new inner subscription |
| 1391 | bool should_continue = false; |
| 1392 | bool should_mark_gen_complete = false; |
| 1393 | bool should_deliver_error = false; |
| 1394 | bool source_exhausted = maybe_next.ok() && IsIterationEnd(*maybe_next); |
| 1395 | Future<T> error_sink; |
| 1396 | { |
| 1397 | auto guard = state->mutex.Lock(); |
| 1398 | if (!maybe_next.ok() || source_exhausted || state->broken) { |
| 1399 | // If here then we will not pull any more from the outer source |
| 1400 | if (!state->broken && !maybe_next.ok()) { |
| 1401 | state->SignalErrorUnlocked(guard); |
| 1402 | // If here then we are the first error so we need to deliver it |
| 1403 | should_deliver_error = true; |
| 1404 | if (!state->waiting_jobs.empty()) { |
| 1405 | error_sink = std::move(*state->waiting_jobs.front()); |
| 1406 | state->waiting_jobs.pop_front(); |
| 1407 | } |
| 1408 | } |
| 1409 | if (source_exhausted) { |
| 1410 | state->source_exhausted = true; |
| 1411 | state->num_running_subscriptions--; |
| 1412 | } |
| 1413 | if (state->MarkTaskFinishedUnlocked(guard)) { |
| 1414 | should_mark_gen_complete = true; |
| 1415 | } |
| 1416 | } else { |
| 1417 | state->active_subscriptions[index] = *maybe_next; |
| 1418 | should_continue = true; |
| 1419 | } |
| 1420 | } |
| 1421 | if (should_deliver_error) { |
| 1422 | state->MarkFinalError(maybe_next.status(), std::move(error_sink)); |
| 1423 | } |
| 1424 | if (should_mark_gen_complete) { |
| 1425 | state->MarkFinishedAndPurge(); |
| 1426 | } |
| 1427 | if (should_continue) { |
| 1428 | // There is a possibility that a large sequence of immediately available inner |
| 1429 | // callbacks could lead to a stack overflow. To avoid this we need to |
| 1430 | // synchronously loop through inner/outer callbacks until we either find an |
| 1431 | // unfinished future or we find an actual item to deliver. |
| 1432 | Future<T> next_item = (*maybe_next)(); |
| 1433 | if (!next_item.TryAddCallback([this] { return InnerCallback(state, index); })) { |
| 1434 | // By setting recursive to true we signal to the inner callback that, if it is |
| 1435 | // empty, instead of adding a new outer callback, it should just immediately |
| 1436 | // return, flagging was_empty so that we know we need to check the next |
| 1437 | // subscription. |
| 1438 | InnerCallback immediate_inner(state, index, /*recursive=*/true); |
| 1439 | immediate_inner(next_item.result()); |
| 1440 | if (immediate_inner.was_empty) { |
| 1441 | Future<AsyncGenerator<T>> next_source = state->PullSource(); |
| 1442 | if (next_source.TryAddCallback( |
| 1443 | [this] { return OuterCallback{state, index}; })) { |
| 1444 | // We hit an unfinished future so we can stop looping |
nothing calls this directly
no test coverage detected