<commit_before>var path = require("path"),
    Setup = require("./setup"),
    ObjectId = require("mongodb").ObjectID,
    fs = require("fs"),
    stream = require('stream'),
    util = require('util'),
    Q = require("q"); 

describe("A suite for stream", function () {

    it("setup", function (done) {
          
        return new Setup().run().then(function(setup) {
        
            var $mongo = setup.$qwebs.resolve("$mongo");
            
            var promises = [];
            
            for (var i = 0; i < 5; i++) {
                var user = {
                    login: "user" + i,
                    password: "password " + i
                };
                
                promises.push($mongo.insert("users", user));
            };
            
            return Q.all(promises);
 
        }).catch(function (error) {
            expect(error.stack).toBeNull();
        }).finally(done);
    });
    
    it("stream", function (done) {
        
        return Q.try(function() {
            var $mongo = setup.$qwebs.resolve("$mongo");
            
            return $mongo.find("users").then(function(cursor) {
                
                var stream = cursor.stream();
     
                stream.once("end", function() {
                    console.log("Stream completed");
                    done();
                });
                
                stream.on("error", function(error) {
                    done();
                    console.log("ERROR");
                });
                
                return stream;
            });
        }).then(function(stream) {
            console.log("Continue to use stream");

        }).catch(function (error) {
            expect(error.stack).toBeNull();
        }).finally();
    });
    
    it("transform stream", function (done) {
        
        return Q.try(function() {
            var $mongo = setup.$qwebs.resolve("$mongo");
            
            return $mongo.find("users").then(function(cursor) {
                return cursor.stream().pipe(new ExtendUser($mongo, {objectMode: true}));
            });
        }).then(function(stream) {
              
            stream.once("finish", function() { //Not end but finish
                console.log("finish");
                done();
            });
        }).catch(function (error) {
            expect(error.stack).toBeNull();
        }).finally();
    });
});

function ExtendUser($mongo, options) {
    stream.Transform.call(this, options);
    this.$mongo = $mongo;
}

util.inherits(ExtendUser, stream.Transform);

ExtendUser.prototype._transform = function (chunk, enc, cb) {
    var self = this;
    self.$mongo.findOne("users", { login: chunk.login }).then(function(user) {
        user.date = new Date();
        self.push(user);
    }).finally(function() {
        cb();
    });
};
<commit_msg>Update stream.spec.js<commit_after>var path = require("path"),
    setup = require("./setup"),
    ObjectId = require("mongodb").ObjectID,
    fs = require("fs"),
    stream = require('stream'),
    util = require('util'),
    Q = require("q"); 

describe("A suite for stream", function () {

    it("setup", function (done) {
          
        return setup.run().then(function() {
        
            var $mongo = setup.$qwebs.resolve("$mongo");
            
            var promises = [];
            
            for (var i = 0; i < 5; i++) {
                var user = {
                    login: "user" + i,
                    password: "password " + i
                };
                
                promises.push($mongo.insert("users", user));
            };
            
            return Q.all(promises);
 
        }).catch(function (error) {
            expect(error.stack).toBeNull();
        }).finally(done);
    });
    
    it("stream", function (done) {
        
        return Q.try(function() {
            var $mongo = setup.$qwebs.resolve("$mongo");
            
            return $mongo.find("users").then(function(cursor) {
                
                var stream = cursor.stream();
     
                stream.once("end", function() {
                    console.log("Stream completed");
                    done();
                });
                
                stream.on("error", function(error) {
                    done();
                    console.log("ERROR");
                });
                
                return stream;
            });
        }).then(function(stream) {
            console.log("Continue to use stream");

        }).catch(function (error) {
            expect(error.stack).toBeNull();
        }).finally();
    });
    
    it("transform stream", function (done) {
        
        return Q.try(function() {
            var $mongo = setup.$qwebs.resolve("$mongo");
            
            return $mongo.find("users").then(function(cursor) {
                return cursor.stream().pipe(new ExtendUser($mongo, {objectMode: true}));
            });
        }).then(function(stream) {
              
            stream.once("finish", function() { //Not end but finish
                console.log("finish");
                done();
            });
        }).catch(function (error) {
            expect(error.stack).toBeNull();
        }).finally();
    });
});

function ExtendUser($mongo, options) {
    stream.Transform.call(this, options);
    this.$mongo = $mongo;
}

util.inherits(ExtendUser, stream.Transform);

ExtendUser.prototype._transform = function (chunk, enc, cb) {
    var self = this;
    self.$mongo.findOne("users", { login: chunk.login }).then(function(user) {
        user.date = new Date();
        self.push(user);
    }).finally(function() {
        cb();
    });
};

{'fraction_non_alphanumeric': 0.15617918772877812, 'fraction_numerical': 0.0006972285166463308, 'mean_word_length': 1.4365180467091294, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5484154', 'n_tokens_mistral': 1561, 'n_tokens_neox': 1500, 'n_words': 286}