GeezerVison is showing github.com — stream: add final method · nodejs/node@07c7f19
File treeExpand 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: doc/api/stream.md +25-3Lines changed: 25 additions & 3 deletions - Display the source diff
- Display the rich diff
| Original file line number | Diff line number | Diff line change |
|---|
@@ -1198,7 +1198,8 @@ on the type of stream being created, as detailed in the chart below: | 1198 | 1198 | <p>[Writable](#stream_class_stream_writable)</p> | 1199 | 1199 | </td> | 1200 | 1200 | <td> | 1201 | | - <p><code>[_write][stream-_write]</code>, <code>[_writev][stream-_writev]</code></p> | | 1201 | + <p><code>[_write][stream-_write]</code>, <code>[_writev][stream-_writev]</code>, | | 1202 | + <code>[_final][stream-_final]</code></p> | 1202 | 1203 | </td> | 1203 | 1204 | </tr> | 1204 | 1205 | <tr> | @@ -1209,7 +1210,8 @@ on the type of stream being created, as detailed in the chart below: | 1209 | 1210 | <p>[Duplex](#stream_class_stream_duplex)</p> | 1210 | 1211 | </td> | 1211 | 1212 | <td> | 1212 | | - <p><code>[_read][stream-_read]</code>, <code>[_write][stream-_write]</code>, <code>[_writev][stream-_writev]</code></p> | | 1213 | + <p><code>[_read][stream-_read]</code>, <code>[_write][stream-_write]</code>, <code>[_writev][stream-_writev]</code>, | | 1214 | + <code>[_final][stream-_final]</code></p> | 1213 | 1215 | </td> | 1214 | 1216 | </tr> | 1215 | 1217 | <tr> | @@ -1220,7 +1222,8 @@ on the type of stream being created, as detailed in the chart below: | 1220 | 1222 | <p>[Transform](#stream_class_stream_transform)</p> | 1221 | 1223 | </td> | 1222 | 1224 | <td> | 1223 | | - <p><code>[_transform][stream-_transform]</code>, <code>[_flush][stream-_flush]</code></p> | | 1225 | + <p><code>[_transform][stream-_transform]</code>, <code>[_flush][stream-_flush]</code>, | | 1226 | + <code>[_final][stream-_final]</code></p> | 1224 | 1227 | </td> | 1225 | 1228 | </tr> | 1226 | 1229 | </table> | @@ -1279,6 +1282,8 @@ constructor and implement the `writable._write()` method. The | 1279 | 1282 | [`stream._writev()`][stream-_writev] method. | 1280 | 1283 | * `destroy` {Function} Implementation for the | 1281 | 1284 | [`stream._destroy()`][writable-_destroy] method. | | 1285 | + * `final` {Function} Implementation for the | | 1286 | + [`stream._final()`][stream-_final] method. | 1282 | 1287 |
| 1283 | 1288 | For example: | 1284 | 1289 |
| @@ -1398,6 +1403,22 @@ added: REPLACEME | 1398 | 1403 | * `callback` {Function} A callback function that takes an optional error argument | 1399 | 1404 | which is invoked when the writable is destroyed. | 1400 | 1405 |
| | 1406 | +#### writable.\_final(callback) | | 1407 | +<!-- YAML | | 1408 | +added: REPLACEME | | 1409 | +--> | | 1410 | + | | 1411 | +* `callback` {Function} Call this function (optionally with an error | | 1412 | + argument) when you are done writing any remaining data. | | 1413 | + | | 1414 | +Note: `_final()` **must not** be called directly. It MAY be implemented | | 1415 | +by child classes, and if so, will be called by the internal Writable | | 1416 | +class methods only. | | 1417 | + | | 1418 | +This optional function will be called before the stream closes, delaying the | | 1419 | +`finish` event until `callback` is called. This is useful to close resources | | 1420 | +or write buffered data before a stream ends. | | 1421 | + | 1401 | 1422 | #### Errors While Writing | 1402 | 1423 |
| 1403 | 1424 | It is recommended that errors occurring during the processing of the | @@ -2115,6 +2136,7 @@ readable buffer so there is nothing for a user to consume. | 2115 | 2136 | [stream-_transform]: #stream_transform_transform_chunk_encoding_callback | 2116 | 2137 | [stream-_write]: #stream_writable_write_chunk_encoding_callback_1 | 2117 | 2138 | [stream-_writev]: #stream_writable_writev_chunks_callback | | 2139 | +[stream-_final]: #stream_writable_final_callback | 2118 | 2140 | [stream-end]: #stream_writable_end_chunk_encoding_callback | 2119 | 2141 | [stream-pause]: #stream_readable_pause | 2120 | 2142 | [stream-push]: #stream_readable_push_chunk_encoding | |
Collapse file Copy file name to clipboardExpand all lines: lib/_stream_writable.js +29-6Lines changed: 29 additions & 6 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -58,6 +58,12 @@ function WritableState(options, stream) { | 58 | 58 | // cast to ints. | 59 | 59 | this.highWaterMark = Math.floor(this.highWaterMark); | 60 | 60 |
| | 61 | + // if _final has been called | | 62 | + this.finalCalled = false; | | 63 | + | | 64 | + // if _final has been called | | 65 | + this.finalCalled = false; | | 66 | + | 61 | 67 | // drain event flag. | 62 | 68 | this.needDrain = false; | 63 | 69 | // at the start of calling end() | @@ -199,6 +205,9 @@ function Writable(options) { | 199 | 205 |
| 200 | 206 | if (typeof options.destroy === 'function') | 201 | 207 | this._destroy = options.destroy; | | 208 | + | | 209 | + if (typeof options.final === 'function') | | 210 | + this._final = options.final; | 202 | 211 | } | 203 | 212 |
| 204 | 213 | Stream.call(this); | @@ -520,23 +529,37 @@ function needFinish(state) { | 520 | 529 | !state.finished && | 521 | 530 | !state.writing); | 522 | 531 | } | 523 | | - | 524 | | -function prefinish(stream, state) { | 525 | | - if (!state.prefinished) { | | 532 | +function callFinal(stream, state) { | | 533 | + stream._final((err) => { | | 534 | + state.pendingcb--; | | 535 | + if (err) { | | 536 | + stream.emit('error', err); | | 537 | + } | 526 | 538 | state.prefinished = true; | 527 | 539 | stream.emit('prefinish'); | | 540 | + finishMaybe(stream, state); | | 541 | + }); | | 542 | +} | | 543 | +function prefinish(stream, state) { | | 544 | + if (!state.prefinished && !state.finalCalled) { | | 545 | + if (typeof stream._final === 'function') { | | 546 | + state.pendingcb++; | | 547 | + state.finalCalled = true; | | 548 | + process.nextTick(callFinal, stream, state); | | 549 | + } else { | | 550 | + state.prefinished = true; | | 551 | + stream.emit('prefinish'); | | 552 | + } | 528 | 553 | } | 529 | 554 | } | 530 | 555 |
| 531 | 556 | function finishMaybe(stream, state) { | 532 | 557 | var need = needFinish(state); | 533 | 558 | if (need) { | | 559 | + prefinish(stream, state); | 534 | 560 | if (state.pendingcb === 0) { | 535 | | - prefinish(stream, state); | 536 | 561 | state.finished = true; | 537 | 562 | stream.emit('finish'); | 538 | | - } else { | 539 | | - prefinish(stream, state); | 540 | 563 | } | 541 | 564 | } | 542 | 565 | return need; | |
Collapse file Copy file name to clipboard +3-11Lines changed: 3 additions & 11 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 1 | 1 | 'use strict'; | 2 | | -require('../common'); | 3 | | -const assert = require('assert'); | | 2 | +const common = require('../common'); | 4 | 3 |
| 5 | 4 | const Readable = require('stream').Readable; | 6 | 5 |
| 7 | | -let _readCalled = false; | 8 | | -function _read(n) { | 9 | | - _readCalled = true; | | 6 | +const _read = common.mustCall(function _read(n) { | 10 | 7 | this.push(null); | 11 | | -} | | 8 | +}); | 12 | 9 |
| 13 | 10 | const r = new Readable({ read: _read }); | 14 | 11 | r.resume(); | 15 | | - | 16 | | -process.on('exit', function() { | 17 | | - assert.strictEqual(r._read, _read); | 18 | | - assert(_readCalled); | 19 | | -}); |
Collapse file Copy file name to clipboard +12-12Lines changed: 12 additions & 12 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 1 | 1 | 'use strict'; | 2 | | -require('../common'); | | 2 | +const common = require('../common'); | 3 | 3 | const assert = require('assert'); | 4 | 4 |
| 5 | 5 | const Transform = require('stream').Transform; | 6 | 6 |
| 7 | | -let _transformCalled = false; | 8 | | -function _transform(d, e, n) { | 9 | | - _transformCalled = true; | | 7 | +const _transform = common.mustCall(function _transform(d, e, n) { | 10 | 8 | n(); | 11 | | -} | | 9 | +}); | 12 | 10 |
| 13 | | -let _flushCalled = false; | 14 | | -function _flush(n) { | 15 | | - _flushCalled = true; | | 11 | +const _final = common.mustCall(function _final(n) { | 16 | 12 | n(); | 17 | | -} | | 13 | +}); | | 14 | + | | 15 | +const _flush = common.mustCall(function _flush(n) { | | 16 | + n(); | | 17 | +}); | 18 | 18 |
| 19 | 19 | const t = new Transform({ | 20 | 20 | transform: _transform, | 21 | | - flush: _flush | | 21 | + flush: _flush, | | 22 | + final: _final | 22 | 23 | }); | 23 | 24 |
| 24 | 25 | const t2 = new Transform({}); | @@ -34,6 +35,5 @@ assert.throws(() => { | 34 | 35 | process.on('exit', () => { | 35 | 36 | assert.strictEqual(t._transform, _transform); | 36 | 37 | assert.strictEqual(t._flush, _flush); | 37 | | - assert.strictEqual(_transformCalled, true); | 38 | | - assert.strictEqual(_flushCalled, true); | | 38 | + assert.strictEqual(t._final, _final); | 39 | 39 | }); |
Collapse file Copy file name to clipboard +100Lines changed: 100 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
| | 1 | +'use strict'; | | 2 | +const common = require('../common'); | | 3 | +const assert = require('assert'); | | 4 | + | | 5 | +const stream = require('stream'); | | 6 | +let state = 0; | | 7 | + | | 8 | +/* | | 9 | +What you do | | 10 | +var stream = new tream.Transform({ | | 11 | + transform: function transformCallback(chunk, _, next) { | | 12 | + // part 1 | | 13 | + this.push(chunk); | | 14 | + //part 2 | | 15 | + next(); | | 16 | + }, | | 17 | + final: function endCallback(done) { | | 18 | + // part 1 | | 19 | + process.nextTick(function () { | | 20 | + // part 2 | | 21 | + done(); | | 22 | + }); | | 23 | + }, | | 24 | + flush: function flushCallback(done) { | | 25 | + // part 1 | | 26 | + process.nextTick(function () { | | 27 | + // part 2 | | 28 | + done(); | | 29 | + }); | | 30 | + } | | 31 | +}); | | 32 | +t.on('data', dataListener); | | 33 | +t.on('end', endListener); | | 34 | +t.on('finish', finishListener); | | 35 | +t.write(1); | | 36 | +t.write(4); | | 37 | +t.end(7, endMethodCallback); | | 38 | +
| | 39 | +The order things are called | | 40 | +
| | 41 | +1. transformCallback part 1 | | 42 | +2. dataListener | | 43 | +3. transformCallback part 2 | | 44 | +4. transformCallback part 1 | | 45 | +5. dataListener | | 46 | +6. transformCallback part 2 | | 47 | +7. transformCallback part 1 | | 48 | +8. dataListener | | 49 | +9. transformCallback part 2 | | 50 | +10. finalCallback part 1 | | 51 | +11. finalCallback part 2 | | 52 | +12. flushCallback part 1 | | 53 | +13. finishListener | | 54 | +14. endMethodCallback | | 55 | +15. flushCallback part 2 | | 56 | +16. endListener | | 57 | +*/ | | 58 | + | | 59 | +const t = new stream.Transform({ | | 60 | + objectMode: true, | | 61 | + transform: common.mustCall(function(chunk, _, next) { | | 62 | + assert.strictEqual(++state, chunk, 'transformCallback part 1'); | | 63 | + this.push(state); | | 64 | + assert.strictEqual(++state, chunk + 2, 'transformCallback part 2'); | | 65 | + process.nextTick(next); | | 66 | + }, 3), | | 67 | + final: common.mustCall(function(done) { | | 68 | + state++; | | 69 | + assert.strictEqual(state, 10, 'finalCallback part 1'); | | 70 | + state++; | | 71 | + assert.strictEqual(state, 11, 'finalCallback part 2'); | | 72 | + done(); | | 73 | + }, 1), | | 74 | + flush: common.mustCall(function(done) { | | 75 | + state++; | | 76 | + assert.strictEqual(state, 12, 'flushCallback part 1'); | | 77 | + process.nextTick(function() { | | 78 | + state++; | | 79 | + assert.strictEqual(state, 15, 'flushCallback part 2'); | | 80 | + done(); | | 81 | + }); | | 82 | + }, 1) | | 83 | +}); | | 84 | +t.on('finish', common.mustCall(function() { | | 85 | + state++; | | 86 | + assert.strictEqual(state, 13, 'finishListener'); | | 87 | +}, 1)); | | 88 | +t.on('end', common.mustCall(function() { | | 89 | + state++; | | 90 | + assert.strictEqual(state, 16, 'end event'); | | 91 | +}, 1)); | | 92 | +t.on('data', common.mustCall(function(d) { | | 93 | + assert.strictEqual(++state, d + 1, 'dataListener'); | | 94 | +}, 3)); | | 95 | +t.write(1); | | 96 | +t.write(4); | | 97 | +t.end(7, common.mustCall(function() { | | 98 | + state++; | | 99 | + assert.strictEqual(state, 14, 'endMethodCallback'); | | 100 | +}, 1)); |
|