source: frontend/node_modules/readable-stream/lib/_stream_writable.js

Last change on this file was 9af201e, checked in by MBK <marija.karapandzova@…>, 11 days ago

Fix frontend appearance

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