GeezerVison is showing github.com — stream: fix regression on duplex end · nodejs/node@1cefb7e
File tree
Expand file treeCollapse file tree
Open diff view settings
Expand file treeCollapse file tree
Open diff view settings
Collapse file
Copy file name to clipboardExpand all lines: lib/internal/streams/writable.js +6-6Lines changed: 6 additions & 6 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -336,6 +336,12 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) { |
336 | 336 |
|
337 | 337 | state.length += len; |
338 | 338 |
|
| 339 | + // stream._write resets state.length |
| 340 | + const ret = state.length < state.highWaterMark; |
| 341 | + // We must ensure that previous needDrain will not be reset to false. |
| 342 | + if (!ret) |
| 343 | + state.needDrain = true; |
| 344 | + |
339 | 345 | if (state.writing || state.corked || state.errored) { |
340 | 346 | state.buffered.push({ chunk, encoding, callback }); |
341 | 347 | if (state.allBuffers && encoding !== 'buffer') { |
@@ -353,12 +359,6 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) { |
353 | 359 | state.sync = false; |
354 | 360 | } |
355 | 361 |
|
356 | | - const ret = state.length < state.highWaterMark; |
357 | | - |
358 | | - // We must ensure that previous needDrain will not be reset to false. |
359 | | - if (!ret) |
360 | | - state.needDrain = true; |
361 | | - |
362 | 362 | // Return false if errored or destroyed in order to break |
363 | 363 | // any synchronous while(stream.write(data)) loops. |
364 | 364 | return ret && !state.errored && !state.destroyed; |
|
Collapse file
Copy file name to clipboard +32Lines changed: 32 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
|
| 1 | +'use strict'; |
| 2 | +// https://github.com/nodejs/node/issues/35926 |
| 3 | +require('../common'); |
| 4 | +const assert = require('assert'); |
| 5 | +const stream = require('stream'); |
| 6 | + |
| 7 | +let loops = 5; |
| 8 | + |
| 9 | +const src = new stream.Readable({ |
| 10 | + read() { |
| 11 | + if (loops--) |
| 12 | + this.push(Buffer.alloc(20000)); |
| 13 | + } |
| 14 | +}); |
| 15 | + |
| 16 | +const dst = new stream.Transform({ |
| 17 | + transform(chunk, output, fn) { |
| 18 | + this.push(null); |
| 19 | + fn(); |
| 20 | + } |
| 21 | +}); |
| 22 | + |
| 23 | +src.pipe(dst); |
| 24 | + |
| 25 | +function parser_end() { |
| 26 | + assert.ok(loops > 0); |
| 27 | + dst.removeAllListeners(); |
| 28 | +} |
| 29 | + |
| 30 | +dst.on('data', () => { }); |
| 31 | +dst.on('end', parser_end); |
| 32 | +dst.on('error', parser_end); |