_stream_writable.js 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697
  1. // Copyright Joyent, Inc. and other Node contributors.
  2. //
  3. // Permission is hereby granted, free of charge, to any person obtaining a
  4. // copy of this software and associated documentation files (the
  5. // "Software"), to deal in the Software without restriction, including
  6. // without limitation the rights to use, copy, modify, merge, publish,
  7. // distribute, sublicense, and/or sell copies of the Software, and to permit
  8. // persons to whom the Software is furnished to do so, subject to the
  9. // following conditions:
  10. //
  11. // The above copyright notice and this permission notice shall be included
  12. // in all copies or substantial portions of the Software.
  13. //
  14. // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
  15. // OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  16. // MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
  17. // NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
  18. // DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
  19. // OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
  20. // USE OR OTHER DEALINGS IN THE SOFTWARE.
  21. // A bit simpler than readable streams.
  22. // Implement an async ._write(chunk, encoding, cb), and it'll handle all
  23. // the drain event emission and buffering.
  24. 'use strict';
  25. module.exports = Writable;
  26. /* <replacement> */
  27. function WriteReq(chunk, encoding, cb) {
  28. this.chunk = chunk;
  29. this.encoding = encoding;
  30. this.callback = cb;
  31. this.next = null;
  32. } // It seems a linked list but it is not
  33. // there will be only 2 of these for each stream
  34. function CorkedRequest(state) {
  35. var _this = this;
  36. this.next = null;
  37. this.entry = null;
  38. this.finish = function () {
  39. onCorkedFinish(_this, state);
  40. };
  41. }
  42. /* </replacement> */
  43. /*<replacement>*/
  44. var Duplex;
  45. /*</replacement>*/
  46. Writable.WritableState = WritableState;
  47. /*<replacement>*/
  48. var internalUtil = {
  49. deprecate: require('util-deprecate')
  50. };
  51. /*</replacement>*/
  52. /*<replacement>*/
  53. var Stream = require('./internal/streams/stream');
  54. /*</replacement>*/
  55. var Buffer = require('buffer').Buffer;
  56. var OurUint8Array = global.Uint8Array || function () {};
  57. function _uint8ArrayToBuffer(chunk) {
  58. return Buffer.from(chunk);
  59. }
  60. function _isUint8Array(obj) {
  61. return Buffer.isBuffer(obj) || obj instanceof OurUint8Array;
  62. }
  63. var destroyImpl = require('./internal/streams/destroy');
  64. var _require = require('./internal/streams/state'),
  65. getHighWaterMark = _require.getHighWaterMark;
  66. var _require$codes = require('../errors').codes,
  67. ERR_INVALID_ARG_TYPE = _require$codes.ERR_INVALID_ARG_TYPE,
  68. ERR_METHOD_NOT_IMPLEMENTED = _require$codes.ERR_METHOD_NOT_IMPLEMENTED,
  69. ERR_MULTIPLE_CALLBACK = _require$codes.ERR_MULTIPLE_CALLBACK,
  70. ERR_STREAM_CANNOT_PIPE = _require$codes.ERR_STREAM_CANNOT_PIPE,
  71. ERR_STREAM_DESTROYED = _require$codes.ERR_STREAM_DESTROYED,
  72. ERR_STREAM_NULL_VALUES = _require$codes.ERR_STREAM_NULL_VALUES,
  73. ERR_STREAM_WRITE_AFTER_END = _require$codes.ERR_STREAM_WRITE_AFTER_END,
  74. ERR_UNKNOWN_ENCODING = _require$codes.ERR_UNKNOWN_ENCODING;
  75. var errorOrDestroy = destroyImpl.errorOrDestroy;
  76. require('inherits')(Writable, Stream);
  77. function nop() {}
  78. function WritableState(options, stream, isDuplex) {
  79. Duplex = Duplex || require('./_stream_duplex');
  80. options = options || {}; // Duplex streams are both readable and writable, but share
  81. // the same options object.
  82. // However, some cases require setting options to different
  83. // values for the readable and the writable sides of the duplex stream,
  84. // e.g. options.readableObjectMode vs. options.writableObjectMode, etc.
  85. if (typeof isDuplex !== 'boolean') isDuplex = stream instanceof Duplex; // object stream flag to indicate whether or not this stream
  86. // contains buffers or objects.
  87. this.objectMode = !!options.objectMode;
  88. if (isDuplex) this.objectMode = this.objectMode || !!options.writableObjectMode; // the point at which write() starts returning false
  89. // Note: 0 is a valid value, means that we always return false if
  90. // the entire buffer is not flushed immediately on write()
  91. this.highWaterMark = getHighWaterMark(this, options, 'writableHighWaterMark', isDuplex); // if _final has been called
  92. this.finalCalled = false; // drain event flag.
  93. this.needDrain = false; // at the start of calling end()
  94. this.ending = false; // when end() has been called, and returned
  95. this.ended = false; // when 'finish' is emitted
  96. this.finished = false; // has it been destroyed
  97. this.destroyed = false; // should we decode strings into buffers before passing to _write?
  98. // this is here so that some node-core streams can optimize string
  99. // handling at a lower level.
  100. var noDecode = options.decodeStrings === false;
  101. this.decodeStrings = !noDecode; // Crypto is kind of old and crusty. Historically, its default string
  102. // encoding is 'binary' so we have to make this configurable.
  103. // Everything else in the universe uses 'utf8', though.
  104. this.defaultEncoding = options.defaultEncoding || 'utf8'; // not an actual buffer we keep track of, but a measurement
  105. // of how much we're waiting to get pushed to some underlying
  106. // socket or file.
  107. this.length = 0; // a flag to see when we're in the middle of a write.
  108. this.writing = false; // when true all writes will be buffered until .uncork() call
  109. this.corked = 0; // a flag to be able to tell if the onwrite cb is called immediately,
  110. // or on a later tick. We set this to true at first, because any
  111. // actions that shouldn't happen until "later" should generally also
  112. // not happen before the first write call.
  113. this.sync = true; // a flag to know if we're processing previously buffered items, which
  114. // may call the _write() callback in the same tick, so that we don't
  115. // end up in an overlapped onwrite situation.
  116. this.bufferProcessing = false; // the callback that's passed to _write(chunk,cb)
  117. this.onwrite = function (er) {
  118. onwrite(stream, er);
  119. }; // the callback that the user supplies to write(chunk,encoding,cb)
  120. this.writecb = null; // the amount that is being written when _write is called.
  121. this.writelen = 0;
  122. this.bufferedRequest = null;
  123. this.lastBufferedRequest = null; // number of pending user-supplied write callbacks
  124. // this must be 0 before 'finish' can be emitted
  125. this.pendingcb = 0; // emit prefinish if the only thing we're waiting for is _write cbs
  126. // This is relevant for synchronous Transform streams
  127. this.prefinished = false; // True if the error was already emitted and should not be thrown again
  128. this.errorEmitted = false; // Should close be emitted on destroy. Defaults to true.
  129. this.emitClose = options.emitClose !== false; // Should .destroy() be called after 'finish' (and potentially 'end')
  130. this.autoDestroy = !!options.autoDestroy; // count buffered requests
  131. this.bufferedRequestCount = 0; // allocate the first CorkedRequest, there is always
  132. // one allocated and free to use, and we maintain at most two
  133. this.corkedRequestsFree = new CorkedRequest(this);
  134. }
  135. WritableState.prototype.getBuffer = function getBuffer() {
  136. var current = this.bufferedRequest;
  137. var out = [];
  138. while (current) {
  139. out.push(current);
  140. current = current.next;
  141. }
  142. return out;
  143. };
  144. (function () {
  145. try {
  146. Object.defineProperty(WritableState.prototype, 'buffer', {
  147. get: internalUtil.deprecate(function writableStateBufferGetter() {
  148. return this.getBuffer();
  149. }, '_writableState.buffer is deprecated. Use _writableState.getBuffer ' + 'instead.', 'DEP0003')
  150. });
  151. } catch (_) {}
  152. })(); // Test _writableState for inheritance to account for Duplex streams,
  153. // whose prototype chain only points to Readable.
  154. var realHasInstance;
  155. if (typeof Symbol === 'function' && Symbol.hasInstance && typeof Function.prototype[Symbol.hasInstance] === 'function') {
  156. realHasInstance = Function.prototype[Symbol.hasInstance];
  157. Object.defineProperty(Writable, Symbol.hasInstance, {
  158. value: function value(object) {
  159. if (realHasInstance.call(this, object)) return true;
  160. if (this !== Writable) return false;
  161. return object && object._writableState instanceof WritableState;
  162. }
  163. });
  164. } else {
  165. realHasInstance = function realHasInstance(object) {
  166. return object instanceof this;
  167. };
  168. }
  169. function Writable(options) {
  170. Duplex = Duplex || require('./_stream_duplex'); // Writable ctor is applied to Duplexes, too.
  171. // `realHasInstance` is necessary because using plain `instanceof`
  172. // would return false, as no `_writableState` property is attached.
  173. // Trying to use the custom `instanceof` for Writable here will also break the
  174. // Node.js LazyTransform implementation, which has a non-trivial getter for
  175. // `_writableState` that would lead to infinite recursion.
  176. // Checking for a Stream.Duplex instance is faster here instead of inside
  177. // the WritableState constructor, at least with V8 6.5
  178. var isDuplex = this instanceof Duplex;
  179. if (!isDuplex && !realHasInstance.call(Writable, this)) return new Writable(options);
  180. this._writableState = new WritableState(options, this, isDuplex); // legacy.
  181. this.writable = true;
  182. if (options) {
  183. if (typeof options.write === 'function') this._write = options.write;
  184. if (typeof options.writev === 'function') this._writev = options.writev;
  185. if (typeof options.destroy === 'function') this._destroy = options.destroy;
  186. if (typeof options.final === 'function') this._final = options.final;
  187. }
  188. Stream.call(this);
  189. } // Otherwise people can pipe Writable streams, which is just wrong.
  190. Writable.prototype.pipe = function () {
  191. errorOrDestroy(this, new ERR_STREAM_CANNOT_PIPE());
  192. };
  193. function writeAfterEnd(stream, cb) {
  194. var er = new ERR_STREAM_WRITE_AFTER_END(); // TODO: defer error events consistently everywhere, not just the cb
  195. errorOrDestroy(stream, er);
  196. process.nextTick(cb, er);
  197. } // Checks that a user-supplied chunk is valid, especially for the particular
  198. // mode the stream is in. Currently this means that `null` is never accepted
  199. // and undefined/non-string values are only allowed in object mode.
  200. function validChunk(stream, state, chunk, cb) {
  201. var er;
  202. if (chunk === null) {
  203. er = new ERR_STREAM_NULL_VALUES();
  204. } else if (typeof chunk !== 'string' && !state.objectMode) {
  205. er = new ERR_INVALID_ARG_TYPE('chunk', ['string', 'Buffer'], chunk);
  206. }
  207. if (er) {
  208. errorOrDestroy(stream, er);
  209. process.nextTick(cb, er);
  210. return false;
  211. }
  212. return true;
  213. }
  214. Writable.prototype.write = function (chunk, encoding, cb) {
  215. var state = this._writableState;
  216. var ret = false;
  217. var isBuf = !state.objectMode && _isUint8Array(chunk);
  218. if (isBuf && !Buffer.isBuffer(chunk)) {
  219. chunk = _uint8ArrayToBuffer(chunk);
  220. }
  221. if (typeof encoding === 'function') {
  222. cb = encoding;
  223. encoding = null;
  224. }
  225. if (isBuf) encoding = 'buffer';else if (!encoding) encoding = state.defaultEncoding;
  226. if (typeof cb !== 'function') cb = nop;
  227. if (state.ending) writeAfterEnd(this, cb);else if (isBuf || validChunk(this, state, chunk, cb)) {
  228. state.pendingcb++;
  229. ret = writeOrBuffer(this, state, isBuf, chunk, encoding, cb);
  230. }
  231. return ret;
  232. };
  233. Writable.prototype.cork = function () {
  234. this._writableState.corked++;
  235. };
  236. Writable.prototype.uncork = function () {
  237. var state = this._writableState;
  238. if (state.corked) {
  239. state.corked--;
  240. if (!state.writing && !state.corked && !state.bufferProcessing && state.bufferedRequest) clearBuffer(this, state);
  241. }
  242. };
  243. Writable.prototype.setDefaultEncoding = function setDefaultEncoding(encoding) {
  244. // node::ParseEncoding() requires lower case.
  245. if (typeof encoding === 'string') encoding = encoding.toLowerCase();
  246. if (!(['hex', 'utf8', 'utf-8', 'ascii', 'binary', 'base64', 'ucs2', 'ucs-2', 'utf16le', 'utf-16le', 'raw'].indexOf((encoding + '').toLowerCase()) > -1)) throw new ERR_UNKNOWN_ENCODING(encoding);
  247. this._writableState.defaultEncoding = encoding;
  248. return this;
  249. };
  250. Object.defineProperty(Writable.prototype, 'writableBuffer', {
  251. // making it explicit this property is not enumerable
  252. // because otherwise some prototype manipulation in
  253. // userland will fail
  254. enumerable: false,
  255. get: function get() {
  256. return this._writableState && this._writableState.getBuffer();
  257. }
  258. });
  259. function decodeChunk(state, chunk, encoding) {
  260. if (!state.objectMode && state.decodeStrings !== false && typeof chunk === 'string') {
  261. chunk = Buffer.from(chunk, encoding);
  262. }
  263. return chunk;
  264. }
  265. Object.defineProperty(Writable.prototype, 'writableHighWaterMark', {
  266. // making it explicit this property is not enumerable
  267. // because otherwise some prototype manipulation in
  268. // userland will fail
  269. enumerable: false,
  270. get: function get() {
  271. return this._writableState.highWaterMark;
  272. }
  273. }); // if we're already writing something, then just put this
  274. // in the queue, and wait our turn. Otherwise, call _write
  275. // If we return false, then we need a drain event, so set that flag.
  276. function writeOrBuffer(stream, state, isBuf, chunk, encoding, cb) {
  277. if (!isBuf) {
  278. var newChunk = decodeChunk(state, chunk, encoding);
  279. if (chunk !== newChunk) {
  280. isBuf = true;
  281. encoding = 'buffer';
  282. chunk = newChunk;
  283. }
  284. }
  285. var len = state.objectMode ? 1 : chunk.length;
  286. state.length += len;
  287. var ret = state.length < state.highWaterMark; // we must ensure that previous needDrain will not be reset to false.
  288. if (!ret) state.needDrain = true;
  289. if (state.writing || state.corked) {
  290. var last = state.lastBufferedRequest;
  291. state.lastBufferedRequest = {
  292. chunk: chunk,
  293. encoding: encoding,
  294. isBuf: isBuf,
  295. callback: cb,
  296. next: null
  297. };
  298. if (last) {
  299. last.next = state.lastBufferedRequest;
  300. } else {
  301. state.bufferedRequest = state.lastBufferedRequest;
  302. }
  303. state.bufferedRequestCount += 1;
  304. } else {
  305. doWrite(stream, state, false, len, chunk, encoding, cb);
  306. }
  307. return ret;
  308. }
  309. function doWrite(stream, state, writev, len, chunk, encoding, cb) {
  310. state.writelen = len;
  311. state.writecb = cb;
  312. state.writing = true;
  313. state.sync = true;
  314. if (state.destroyed) state.onwrite(new ERR_STREAM_DESTROYED('write'));else if (writev) stream._writev(chunk, state.onwrite);else stream._write(chunk, encoding, state.onwrite);
  315. state.sync = false;
  316. }
  317. function onwriteError(stream, state, sync, er, cb) {
  318. --state.pendingcb;
  319. if (sync) {
  320. // defer the callback if we are being called synchronously
  321. // to avoid piling up things on the stack
  322. process.nextTick(cb, er); // this can emit finish, and it will always happen
  323. // after error
  324. process.nextTick(finishMaybe, stream, state);
  325. stream._writableState.errorEmitted = true;
  326. errorOrDestroy(stream, er);
  327. } else {
  328. // the caller expect this to happen before if
  329. // it is async
  330. cb(er);
  331. stream._writableState.errorEmitted = true;
  332. errorOrDestroy(stream, er); // this can emit finish, but finish must
  333. // always follow error
  334. finishMaybe(stream, state);
  335. }
  336. }
  337. function onwriteStateUpdate(state) {
  338. state.writing = false;
  339. state.writecb = null;
  340. state.length -= state.writelen;
  341. state.writelen = 0;
  342. }
  343. function onwrite(stream, er) {
  344. var state = stream._writableState;
  345. var sync = state.sync;
  346. var cb = state.writecb;
  347. if (typeof cb !== 'function') throw new ERR_MULTIPLE_CALLBACK();
  348. onwriteStateUpdate(state);
  349. if (er) onwriteError(stream, state, sync, er, cb);else {
  350. // Check if we're actually ready to finish, but don't emit yet
  351. var finished = needFinish(state) || stream.destroyed;
  352. if (!finished && !state.corked && !state.bufferProcessing && state.bufferedRequest) {
  353. clearBuffer(stream, state);
  354. }
  355. if (sync) {
  356. process.nextTick(afterWrite, stream, state, finished, cb);
  357. } else {
  358. afterWrite(stream, state, finished, cb);
  359. }
  360. }
  361. }
  362. function afterWrite(stream, state, finished, cb) {
  363. if (!finished) onwriteDrain(stream, state);
  364. state.pendingcb--;
  365. cb();
  366. finishMaybe(stream, state);
  367. } // Must force callback to be called on nextTick, so that we don't
  368. // emit 'drain' before the write() consumer gets the 'false' return
  369. // value, and has a chance to attach a 'drain' listener.
  370. function onwriteDrain(stream, state) {
  371. if (state.length === 0 && state.needDrain) {
  372. state.needDrain = false;
  373. stream.emit('drain');
  374. }
  375. } // if there's something in the buffer waiting, then process it
  376. function clearBuffer(stream, state) {
  377. state.bufferProcessing = true;
  378. var entry = state.bufferedRequest;
  379. if (stream._writev && entry && entry.next) {
  380. // Fast case, write everything using _writev()
  381. var l = state.bufferedRequestCount;
  382. var buffer = new Array(l);
  383. var holder = state.corkedRequestsFree;
  384. holder.entry = entry;
  385. var count = 0;
  386. var allBuffers = true;
  387. while (entry) {
  388. buffer[count] = entry;
  389. if (!entry.isBuf) allBuffers = false;
  390. entry = entry.next;
  391. count += 1;
  392. }
  393. buffer.allBuffers = allBuffers;
  394. doWrite(stream, state, true, state.length, buffer, '', holder.finish); // doWrite is almost always async, defer these to save a bit of time
  395. // as the hot path ends with doWrite
  396. state.pendingcb++;
  397. state.lastBufferedRequest = null;
  398. if (holder.next) {
  399. state.corkedRequestsFree = holder.next;
  400. holder.next = null;
  401. } else {
  402. state.corkedRequestsFree = new CorkedRequest(state);
  403. }
  404. state.bufferedRequestCount = 0;
  405. } else {
  406. // Slow case, write chunks one-by-one
  407. while (entry) {
  408. var chunk = entry.chunk;
  409. var encoding = entry.encoding;
  410. var cb = entry.callback;
  411. var len = state.objectMode ? 1 : chunk.length;
  412. doWrite(stream, state, false, len, chunk, encoding, cb);
  413. entry = entry.next;
  414. state.bufferedRequestCount--; // if we didn't call the onwrite immediately, then
  415. // it means that we need to wait until it does.
  416. // also, that means that the chunk and cb are currently
  417. // being processed, so move the buffer counter past them.
  418. if (state.writing) {
  419. break;
  420. }
  421. }
  422. if (entry === null) state.lastBufferedRequest = null;
  423. }
  424. state.bufferedRequest = entry;
  425. state.bufferProcessing = false;
  426. }
  427. Writable.prototype._write = function (chunk, encoding, cb) {
  428. cb(new ERR_METHOD_NOT_IMPLEMENTED('_write()'));
  429. };
  430. Writable.prototype._writev = null;
  431. Writable.prototype.end = function (chunk, encoding, cb) {
  432. var state = this._writableState;
  433. if (typeof chunk === 'function') {
  434. cb = chunk;
  435. chunk = null;
  436. encoding = null;
  437. } else if (typeof encoding === 'function') {
  438. cb = encoding;
  439. encoding = null;
  440. }
  441. if (chunk !== null && chunk !== undefined) this.write(chunk, encoding); // .end() fully uncorks
  442. if (state.corked) {
  443. state.corked = 1;
  444. this.uncork();
  445. } // ignore unnecessary end() calls.
  446. if (!state.ending) endWritable(this, state, cb);
  447. return this;
  448. };
  449. Object.defineProperty(Writable.prototype, 'writableLength', {
  450. // making it explicit this property is not enumerable
  451. // because otherwise some prototype manipulation in
  452. // userland will fail
  453. enumerable: false,
  454. get: function get() {
  455. return this._writableState.length;
  456. }
  457. });
  458. function needFinish(state) {
  459. return state.ending && state.length === 0 && state.bufferedRequest === null && !state.finished && !state.writing;
  460. }
  461. function callFinal(stream, state) {
  462. stream._final(function (err) {
  463. state.pendingcb--;
  464. if (err) {
  465. errorOrDestroy(stream, err);
  466. }
  467. state.prefinished = true;
  468. stream.emit('prefinish');
  469. finishMaybe(stream, state);
  470. });
  471. }
  472. function prefinish(stream, state) {
  473. if (!state.prefinished && !state.finalCalled) {
  474. if (typeof stream._final === 'function' && !state.destroyed) {
  475. state.pendingcb++;
  476. state.finalCalled = true;
  477. process.nextTick(callFinal, stream, state);
  478. } else {
  479. state.prefinished = true;
  480. stream.emit('prefinish');
  481. }
  482. }
  483. }
  484. function finishMaybe(stream, state) {
  485. var need = needFinish(state);
  486. if (need) {
  487. prefinish(stream, state);
  488. if (state.pendingcb === 0) {
  489. state.finished = true;
  490. stream.emit('finish');
  491. if (state.autoDestroy) {
  492. // In case of duplex streams we need a way to detect
  493. // if the readable side is ready for autoDestroy as well
  494. var rState = stream._readableState;
  495. if (!rState || rState.autoDestroy && rState.endEmitted) {
  496. stream.destroy();
  497. }
  498. }
  499. }
  500. }
  501. return need;
  502. }
  503. function endWritable(stream, state, cb) {
  504. state.ending = true;
  505. finishMaybe(stream, state);
  506. if (cb) {
  507. if (state.finished) process.nextTick(cb);else stream.once('finish', cb);
  508. }
  509. state.ended = true;
  510. stream.writable = false;
  511. }
  512. function onCorkedFinish(corkReq, state, err) {
  513. var entry = corkReq.entry;
  514. corkReq.entry = null;
  515. while (entry) {
  516. var cb = entry.callback;
  517. state.pendingcb--;
  518. cb(err);
  519. entry = entry.next;
  520. } // reuse the free corkReq.
  521. state.corkedRequestsFree.next = corkReq;
  522. }
  523. Object.defineProperty(Writable.prototype, 'destroyed', {
  524. // making it explicit this property is not enumerable
  525. // because otherwise some prototype manipulation in
  526. // userland will fail
  527. enumerable: false,
  528. get: function get() {
  529. if (this._writableState === undefined) {
  530. return false;
  531. }
  532. return this._writableState.destroyed;
  533. },
  534. set: function set(value) {
  535. // we ignore the value if the stream
  536. // has not been initialized yet
  537. if (!this._writableState) {
  538. return;
  539. } // backward compatibility, the user is explicitly
  540. // managing destroyed
  541. this._writableState.destroyed = value;
  542. }
  543. });
  544. Writable.prototype.destroy = destroyImpl.destroy;
  545. Writable.prototype._undestroy = destroyImpl.undestroy;
  546. Writable.prototype._destroy = function (err, cb) {
  547. cb(err);
  548. };