| 1490 | } |
| 1491 | |
| 1492 | void AsyncPipeline::endPipes() |
| 1493 | { |
| 1494 | if (endingPipes) |
| 1495 | { |
| 1496 | return; |
| 1497 | } |
| 1498 | if (hasPendingWrites() or (source != nullptr and source->hasQueuedData())) |
| 1499 | { |
| 1500 | shouldEndWhenDrained = true; |
| 1501 | if (not hasPendingWrites() and source != nullptr) |
| 1502 | { |
| 1503 | source->resumeReading(); |
| 1504 | } |
| 1505 | return; |
| 1506 | } |
| 1507 | |
| 1508 | endingPipes = true; |
| 1509 | bool allEnded = true; |
| 1510 | for (AsyncWritableStream* sink : sinks) |
| 1511 | { |
| 1512 | if (sink == nullptr) |
| 1513 | break; |
| 1514 | if (sink->isStillWriting()) |
| 1515 | { |
| 1516 | const bool res = sink->eventFinish.addListener<AsyncPipeline, &AsyncPipeline::afterSinkEnd>(*this); |
| 1517 | SC_ASYNC_STREAMS_ASSERT_RELEASE(res); |
| 1518 | allEnded = false; |
| 1519 | } |
| 1520 | sink->end(); |
| 1521 | } |
| 1522 | if (allEnded) |
| 1523 | { |
| 1524 | SC_ASYNC_STREAMS_ASSERT_RELEASE(unpipe()); |
| 1525 | } |
| 1526 | } |
| 1527 | |
| 1528 | void AsyncPipeline::afterSinkEnd() |
| 1529 | { |
nothing calls this directly
no test coverage detected