| 992 | } |
| 993 | |
| 994 | function pause() { |
| 995 | // If the user unpiped during `dest.write()`, it is possible |
| 996 | // to get stuck in a permanently paused state if that write |
| 997 | // also returned false. |
| 998 | // => Check whether `dest` is still a piping destination. |
| 999 | if (!cleanedUp) { |
| 1000 | if (state.pipes.length === 1 && state.pipes[0] === dest) { |
| 1001 | debug('false write response, pause', 0); |
| 1002 | state.awaitDrainWriters = dest; |
| 1003 | state[kState] &= ~kMultiAwaitDrain; |
| 1004 | } else if (state.pipes.length > 1 && state.pipes.includes(dest)) { |
| 1005 | debug('false write response, pause', state.awaitDrainWriters.size); |
| 1006 | state.awaitDrainWriters.add(dest); |
| 1007 | } |
| 1008 | src.pause(); |
| 1009 | } |
| 1010 | if (!ondrain) { |
| 1011 | // When the dest drains, it reduces the awaitDrain counter |
| 1012 | // on the source. This would be more elegant with a .once() |
| 1013 | // handler in flow(), but adding and removing repeatedly is |
| 1014 | // too slow. |
| 1015 | ondrain = pipeOnDrain(src, dest); |
| 1016 | dest.on('drain', ondrain); |
| 1017 | } |
| 1018 | } |
| 1019 | |
| 1020 | src.on('data', ondata); |
| 1021 | function ondata(chunk) { |