writer.js

var Q = require("q");

/**
 * Wraps a Node writable stream, providing an API similar to
 * Narwhal's synchronous `io` streams, except returning and
 * accepting promises for long-latency operations.
 *
 * @param stream any Node writable stream
 * @returns {Promise * Writer} a promise for the
 * text writer.
 */
module.exports = Writer;

var version = process.versions.node.split('.');
var supportsFinish = version[0] >= 0 && version[1] >= 10;

function Writer(_stream, charset) {
    var self = Object.create(Writer.prototype);

    if (charset && _stream.setEncoding) // TODO complain about inconsistency
        _stream.setEncoding(charset);

    var drained = Q.defer();

    _stream.on("error", function (reason) {
        drained.reject(reason);
        drained = Q.defer();
    });

    _stream.on("drain", function () {
        drained.resolve();
        drained = Q.defer();
    });

    /***
     * Writes content to the stream.
     * @param {String} content
     * @returns {Promise * Undefined} a promise that will
     * be resolved when the buffer is empty, meaning
     * that all of the content has been sent.
     */
    self.write = function (content) {
        if (!_stream.writeable && !_stream.writable)
            return Q.reject(new Error("Can't write to non-writable (possibly closed) stream"));
        if (typeof content !== "string") {
            content = new Buffer(content);
        }
        if (!_stream.write(content)) {
            return drained.promise;
        } else {
            return Q.resolve();
        }
    };

    /***
     * Waits for all data to flush on the stream.
     *
     * @returns {Promise * Undefined} a promise that will
     * be resolved when the buffer is empty
     */
    self.flush = function () {
        return drained.promise;
    };

    /***
     * Closes the stream, waiting for the internal buffer
     * to flush.
     *
     * @returns {Promise * Undefined} a promise that will
     * be resolved when the stream has finished writing,
     * flushing, and closed.
     */
    self.close = function () {
        var finished;

        if (supportsFinish) { // new Streams, listen for `finish` event
            finished = Q.defer();
            _stream.on("finish", function () {
                finished.resolve();
            });
            _stream.on("error", function (reason) {
                finished.reject(reason);
            });
        }

        _stream.end();
        drained.resolve(); // we will get no further drain events
        if (finished) { // closing not explicitly observable
            return finished.promise;
        } else {
            return Q(); // just resolve for old Streams
        }
    };

    /***
     * Terminates writing on a stream, closing before
     * the internal buffer drains.
     *
     * @returns {Promise * Undefined} a promise that will
     * be resolved when the stream has finished closing.
     */
    self.destroy = function () {
        _stream.destroy();
        drained.resolve(); // we will get no further drain events
        return Q.resolve(); // destruction not explicitly observable
    };

    return Q(self); // todo returns the begin.promise
}