source: node_modules/end-of-stream/index.js@ 81bc7da

var once = require('once');

var noop = function() {};

var qnt = global.Bare ? queueMicrotask : process.nextTick.bind(process);

var isRequest = function(stream) {
        return stream.setHeader && typeof stream.abort === 'function';
};

var isChildProcess = function(stream) {
        return stream.stdio && Array.isArray(stream.stdio) && stream.stdio.length === 3
};

var eos = function(stream, opts, callback) {
        if (typeof opts === 'function') return eos(stream, null, opts);
        if (!opts) opts = {};

        callback = once(callback || noop);

        var ws = stream._writableState;
        var rs = stream._readableState;
        var readable = opts.readable || (opts.readable !== false && stream.readable);
        var writable = opts.writable || (opts.writable !== false && stream.writable);
        var cancelled = false;

        var onlegacyfinish = function() {
                if (!stream.writable) onfinish();
        };

        var onfinish = function() {
                writable = false;
                if (!readable) callback.call(stream);
        };

        var onend = function() {
                readable = false;
                if (!writable) callback.call(stream);
        };

        var onexit = function(exitCode) {
                callback.call(stream, exitCode ? new Error('exited with error code: ' + exitCode) : null);
        };

        var onerror = function(err) {
                callback.call(stream, err);
        };

        var onclose = function() {
                qnt(onclosenexttick);
        };

        var onclosenexttick = function() {
                if (cancelled) return;
                if (readable && !(rs && (rs.ended && !rs.destroyed))) return callback.call(stream, new Error('premature close'));
                if (writable && !(ws && (ws.ended && !ws.destroyed))) return callback.call(stream, new Error('premature close'));
        };

        var onrequest = function() {
                stream.req.on('finish', onfinish);
        };

        if (isRequest(stream)) {
                stream.on('complete', onfinish);
                stream.on('abort', onclose);
                if (stream.req) onrequest();
                else stream.on('request', onrequest);
        } else if (writable && !ws) { // legacy streams
                stream.on('end', onlegacyfinish);
                stream.on('close', onlegacyfinish);
        }

        if (isChildProcess(stream)) stream.on('exit', onexit);

        stream.on('end', onend);
        stream.on('finish', onfinish);
        if (opts.error !== false) stream.on('error', onerror);
        stream.on('close', onclose);

        return function() {
                cancelled = true;
                stream.removeListener('complete', onfinish);
                stream.removeListener('abort', onclose);
                stream.removeListener('request', onrequest);
                if (stream.req) stream.req.removeListener('finish', onfinish);
                stream.removeListener('end', onlegacyfinish);
                stream.removeListener('close', onlegacyfinish);
                stream.removeListener('finish', onfinish);
                stream.removeListener('exit', onexit);
                stream.removeListener('end', onend);
                stream.removeListener('error', onerror);
                stream.removeListener('close', onclose);
        };
};

module.exports = eos;