| 303 | std::move(is_next), std::move(initial_value))) {} |
| 304 | |
| 305 | Future<T> operator()() { |
| 306 | { |
| 307 | auto guard = state_->mutex.Lock(); |
| 308 | // We can send a result immediately if the top of the queue is either an |
| 309 | // error or the next item |
| 310 | if (!state_->queue.empty() && |
| 311 | (!state_->queue.top().ok() || |
| 312 | state_->is_next(state_->previous_value, *state_->queue.top()))) { |
| 313 | auto result = std::move(state_->queue.top()); |
| 314 | if (result.ok()) { |
| 315 | state_->previous_value = *result; |
| 316 | } |
| 317 | state_->queue.pop(); |
| 318 | return Future<T>::MakeFinished(result); |
| 319 | } |
| 320 | if (state_->finished) { |
| 321 | return AsyncGeneratorEnd<T>(); |
| 322 | } |
| 323 | // The next item is not in the queue so we will need to wait |
| 324 | auto new_waiting_fut = Future<T>::Make(); |
| 325 | state_->waiting_future = new_waiting_fut; |
| 326 | guard.Unlock(); |
| 327 | state_->source().AddCallback(Callback{state_}); |
| 328 | return new_waiting_fut; |
| 329 | } |
| 330 | } |
| 331 | |
| 332 | private: |
| 333 | struct WrappedComesAfter { |