// A bit simpler than readable streams. // Implement an async ._write(chunk, encoding, cb), and it'll handle all // the drain event emission and buffering.

'use strict';

module.exports = Writable;

/*<replacement>*/ var processNextTick = require('process-nextick-args'); /*</replacement>*/

/*<replacement>*/ var asyncWrite = !process.browser && ['v0.10', 'v0.9.'].indexOf(process.version.slice(0, 5)) > -1 ? setImmediate : processNextTick; /*</replacement>*/

/*<replacement>*/ var Buffer = require('buffer').Buffer; /*</replacement>*/

Writable.WritableState = WritableState;

/*<replacement>*/ var util = require('core-util-is'); util.inherits = require('inherits'); /*</replacement>*/

/*<replacement>*/ var internalUtil = {

deprecate: require('util-deprecate')

}; /*</replacement>*/

/*<replacement>*/ var Stream; (function () {

try {
  Stream = require('st' + 'ream');
} catch (_) {} finally {
  if (!Stream) Stream = require('events').EventEmitter;
}

})(); /*</replacement>*/

var Buffer = require('buffer').Buffer;

util.inherits(Writable, Stream);

function nop() {}

function WriteReq(chunk, encoding, cb) {

this.chunk = chunk;
this.encoding = encoding;
this.callback = cb;
this.next = null;

}

var Duplex; function WritableState(options, stream) {

Duplex = Duplex || require('./_stream_duplex');

options = options || {};

// object stream flag to indicate whether or not this stream
// contains buffers or objects.
this.objectMode = !!options.objectMode;

if (stream instanceof Duplex) this.objectMode = this.objectMode || !!options.writableObjectMode;

// the point at which write() starts returning false
// Note: 0 is a valid value, means that we always return false if
// the entire buffer is not flushed immediately on write()
var hwm = options.highWaterMark;
var defaultHwm = this.objectMode ? 16 : 16 * 1024;
this.highWaterMark = hwm || hwm === 0 ? hwm : defaultHwm;

// cast to ints.
this.highWaterMark = ~ ~this.highWaterMark;

this.needDrain = false;
// at the start of calling end()
this.ending = false;
// when end() has been called, and returned
this.ended = false;
// when 'finish' is emitted
this.finished = false;

// should we decode strings into buffers before passing to _write?
// this is here so that some node-core streams can optimize string
// handling at a lower level.
var noDecode = options.decodeStrings === false;
this.decodeStrings = !noDecode;

// Crypto is kind of old and crusty.  Historically, its default string
// encoding is 'binary' so we have to make this configurable.
// Everything else in the universe uses 'utf8', though.
this.defaultEncoding = options.defaultEncoding || 'utf8';

// not an actual buffer we keep track of, but a measurement
// of how much we're waiting to get pushed to some underlying
// socket or file.
this.length = 0;

// a flag to see when we're in the middle of a write.
this.writing = false;

// when true all writes will be buffered until .uncork() call
this.corked = 0;

// a flag to be able to tell if the onwrite cb is called immediately,
// or on a later tick.  We set this to true at first, because any
// actions that shouldn't happen until "later" should generally also
// not happen before the first write call.
this.sync = true;

// a flag to know if we're processing previously buffered items, which
// may call the _write() callback in the same tick, so that we don't
// end up in an overlapped onwrite situation.
this.bufferProcessing = false;

// the callback that's passed to _write(chunk,cb)
this.onwrite = function (er) {
  onwrite(stream, er);
};

// the callback that the user supplies to write(chunk,encoding,cb)
this.writecb = null;

// the amount that is being written when _write is called.
this.writelen = 0;

this.bufferedRequest = null;
this.lastBufferedRequest = null;

// number of pending user-supplied write callbacks
// this must be 0 before 'finish' can be emitted
this.pendingcb = 0;

// emit prefinish if the only thing we're waiting for is _write cbs
// This is relevant for synchronous Transform streams
this.prefinished = false;

// True if the error was already emitted and should not be thrown again
this.errorEmitted = false;

// count buffered requests
this.bufferedRequestCount = 0;

// create the two objects needed to store the corked requests
// they are not a linked list, as no new elements are inserted in there
this.corkedRequestsFree = new CorkedRequest(this);
this.corkedRequestsFree.next = new CorkedRequest(this);

}

WritableState.prototype.getBuffer = function writableStateGetBuffer() {

var current = this.bufferedRequest;
var out = [];
while (current) {
  out.push(current);
  current = current.next;
}
return out;

};

(function () {

try {
  Object.defineProperty(WritableState.prototype, 'buffer', {
    get: internalUtil.deprecate(function () {
      return this.getBuffer();
    }, '_writableState.buffer is deprecated. Use _writableState.getBuffer ' + 'instead.')
  });
} catch (_) {}

})();

var Duplex; function Writable(options) {

Duplex = Duplex || require('./_stream_duplex');

// Writable ctor is applied to Duplexes, though they're not
// instanceof Writable, they're instanceof Readable.
if (!(this instanceof Writable) && !(this instanceof Duplex)) return new Writable(options);

this._writableState = new WritableState(options, this);

// legacy.
this.writable = true;

if (options) {
  if (typeof options.write === 'function') this._write = options.write;

  if (typeof options.writev === 'function') this._writev = options.writev;
}

Stream.call(this);

}

// Otherwise people can pipe Writable streams, which is just wrong. Writable.prototype.pipe = function () {

this.emit('error', new Error('Cannot pipe. Not readable.'));

};

function writeAfterEnd(stream, cb) {

var er = new Error('write after end');
// TODO: defer error events consistently everywhere, not just the cb
stream.emit('error', er);
processNextTick(cb, er);

}

// If we get something that is not a buffer, string, null, or undefined, // and we're not in objectMode, then that's an error. // Otherwise stream chunks are all considered to be of length=1, and the // watermarks determine how many objects to keep in the buffer, rather than // how many bytes or characters. function validChunk(stream, state, chunk, cb) {

var valid = true;

if (!Buffer.isBuffer(chunk) && typeof chunk !== 'string' && chunk !== null && chunk !== undefined && !state.objectMode) {
  var er = new TypeError('Invalid non-string/buffer chunk');
  stream.emit('error', er);
  processNextTick(cb, er);
  valid = false;
}
return valid;

}

Writable.prototype.write = function (chunk, encoding, cb) {

var state = this._writableState;
var ret = false;

if (typeof encoding === 'function') {
  cb = encoding;
  encoding = null;
}

if (Buffer.isBuffer(chunk)) encoding = 'buffer';else if (!encoding) encoding = state.defaultEncoding;

if (typeof cb !== 'function') cb = nop;

if (state.ended) writeAfterEnd(this, cb);else if (validChunk(this, state, chunk, cb)) {
  state.pendingcb++;
  ret = writeOrBuffer(this, state, chunk, encoding, cb);
}

return ret;

};

Writable.prototype.cork = function () {

var state = this._writableState;

state.corked++;

};

Writable.prototype.uncork = function () {

var state = this._writableState;

if (state.corked) {
  state.corked--;

  if (!state.writing && !state.corked && !state.finished && !state.bufferProcessing && state.bufferedRequest) clearBuffer(this, state);
}

};

Writable.prototype.setDefaultEncoding = function setDefaultEncoding(encoding) {

// node::ParseEncoding() requires lower case.
if (typeof encoding === 'string') encoding = encoding.toLowerCase();
if (!(['hex', 'utf8', 'utf-8', 'ascii', 'binary', 'base64', 'ucs2', 'ucs-2', 'utf16le', 'utf-16le', 'raw'].indexOf((encoding + '').toLowerCase()) > -1)) throw new TypeError('Unknown encoding: ' + encoding);
this._writableState.defaultEncoding = encoding;

};

function decodeChunk(state, chunk, encoding) {

if (!state.objectMode && state.decodeStrings !== false && typeof chunk === 'string') {
  chunk = new Buffer(chunk, encoding);
}
return chunk;

}

// if we're already writing something, then just put this // in the queue, and wait our turn. Otherwise, call _write // If we return false, then we need a drain event, so set that flag. function writeOrBuffer(stream, state, chunk, encoding, cb) {

chunk = decodeChunk(state, chunk, encoding);

if (Buffer.isBuffer(chunk)) encoding = 'buffer';
var len = state.objectMode ? 1 : chunk.length;

state.length += len;

var ret = state.length < state.highWaterMark;
// we must ensure that previous needDrain will not be reset to false.
if (!ret) state.needDrain = true;

if (state.writing || state.corked) {
  var last = state.lastBufferedRequest;
  state.lastBufferedRequest = new WriteReq(chunk, encoding, cb);
  if (last) {
    last.next = state.lastBufferedRequest;
  } else {
    state.bufferedRequest = state.lastBufferedRequest;
  }
  state.bufferedRequestCount += 1;
} else {
  doWrite(stream, state, false, len, chunk, encoding, cb);
}

return ret;

}

function doWrite(stream, state, writev, len, chunk, encoding, cb) {

state.writelen = len;
state.writecb = cb;
state.writing = true;
state.sync = true;
if (writev) stream._writev(chunk, state.onwrite);else stream._write(chunk, encoding, state.onwrite);
state.sync = false;

}

function onwriteError(stream, state, sync, er, cb) {

--state.pendingcb;
if (sync) processNextTick(cb, er);else cb(er);

stream._writableState.errorEmitted = true;
stream.emit('error', er);

}

function onwriteStateUpdate(state) {

state.writing = false;
state.writecb = null;
state.length -= state.writelen;
state.writelen = 0;

}

function onwrite(stream, er) {

var state = stream._writableState;
var sync = state.sync;
var cb = state.writecb;

onwriteStateUpdate(state);

if (er) onwriteError(stream, state, sync, er, cb);else {
  // Check if we're actually ready to finish, but don't emit yet
  var finished = needFinish(state);

  if (!finished && !state.corked && !state.bufferProcessing && state.bufferedRequest) {
    clearBuffer(stream, state);
  }

  if (sync) {
    /*<replacement>*/
    asyncWrite(afterWrite, stream, state, finished, cb);
    /*</replacement>*/
  } else {
      afterWrite(stream, state, finished, cb);
    }
}

}

function afterWrite(stream, state, finished, cb) {

if (!finished) onwriteDrain(stream, state);
state.pendingcb--;
cb();
finishMaybe(stream, state);

}

// Must force callback to be called on nextTick, so that we don't // emit 'drain' before the write() consumer gets the 'false' return // value, and has a chance to attach a 'drain' listener. function onwriteDrain(stream, state) {

if (state.length === 0 && state.needDrain) {
  state.needDrain = false;
  stream.emit('drain');
}

}

// if there's something in the buffer waiting, then process it function clearBuffer(stream, state) {

state.bufferProcessing = true;
var entry = state.bufferedRequest;

if (stream._writev && entry && entry.next) {
  // Fast case, write everything using _writev()
  var l = state.bufferedRequestCount;
  var buffer = new Array(l);
  var holder = state.corkedRequestsFree;
  holder.entry = entry;

  var count = 0;
  while (entry) {
    buffer[count] = entry;
    entry = entry.next;
    count += 1;
  }

  doWrite(stream, state, true, state.length, buffer, '', holder.finish);

  // doWrite is always async, defer these to save a bit of time
  // as the hot path ends with doWrite
  state.pendingcb++;
  state.lastBufferedRequest = null;
  state.corkedRequestsFree = holder.next;
  holder.next = null;
} else {
  // Slow case, write chunks one-by-one
  while (entry) {
    var chunk = entry.chunk;
    var encoding = entry.encoding;
    var cb = entry.callback;
    var len = state.objectMode ? 1 : chunk.length;

    doWrite(stream, state, false, len, chunk, encoding, cb);
    entry = entry.next;
    // if we didn't call the onwrite immediately, then
    // it means that we need to wait until it does.
    // also, that means that the chunk and cb are currently
    // being processed, so move the buffer counter past them.
    if (state.writing) {
      break;
    }
  }

  if (entry === null) state.lastBufferedRequest = null;
}

state.bufferedRequestCount = 0;
state.bufferedRequest = entry;
state.bufferProcessing = false;

}

Writable.prototype._write = function (chunk, encoding, cb) {

cb(new Error('not implemented'));

};

Writable.prototype._writev = null;

Writable.prototype.end = function (chunk, encoding, cb) {

var state = this._writableState;

if (typeof chunk === 'function') {
  cb = chunk;
  chunk = null;
  encoding = null;
} else if (typeof encoding === 'function') {
  cb = encoding;
  encoding = null;
}

if (chunk !== null && chunk !== undefined) this.write(chunk, encoding);

// .end() fully uncorks
if (state.corked) {
  state.corked = 1;
  this.uncork();
}

// ignore unnecessary end() calls.
if (!state.ending && !state.finished) endWritable(this, state, cb);

};

function needFinish(state) {

return state.ending && state.length === 0 && state.bufferedRequest === null && !state.finished && !state.writing;

}

function prefinish(stream, state) {

if (!state.prefinished) {
  state.prefinished = true;
  stream.emit('prefinish');
}

}

function finishMaybe(stream, state) {

var need = needFinish(state);
if (need) {
  if (state.pendingcb === 0) {
    prefinish(stream, state);
    state.finished = true;
    stream.emit('finish');
  } else {
    prefinish(stream, state);
  }
}
return need;

}

function endWritable(stream, state, cb) {

state.ending = true;
finishMaybe(stream, state);
if (cb) {
  if (state.finished) processNextTick(cb);else stream.once('finish', cb);
}
state.ended = true;
stream.writable = false;

}

// It seems a linked list but it is not // there will be only 2 of these for each stream function CorkedRequest(state) {

var _this = this;

this.next = null;
this.entry = null;

this.finish = function (err) {
  var entry = _this.entry;
  _this.entry = null;
  while (entry) {
    var cb = entry.callback;
    state.pendingcb--;
    cb(err);
    entry = entry.next;
  }
  if (state.corkedRequestsFree) {
    state.corkedRequestsFree.next = _this;
  } else {
    state.corkedRequestsFree = _this;
  }
};

}