All files / Nodejs/test bufferTest.js

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
const ls = require("../").streams;
const async = require("async");
 
const PassThrough = require("stream").PassThrough;
 
describe("streams", function() {
	describe("buffer", function() {
		this.timeout(1000 * 60);
		it("should flush on timer", (done) => {
			let pass = new PassThrough({
				objectMode: true
			});
			ls.pipe(pass, ls.buffer({
				time: {
					seconds: 1
				},
				writeStream: true
			}, (obj, done) => {
				i++;
				done(null, obj)
			}, (done) => {
				console.log("I saw ", i, " events");
				i = 0;
				done();
			}), (err) => {
				console.log("There was an error: ", err);
				console.log("all done");
				done();
			});
			let i = 0;
 
			let x = 0;
			async.doWhilst((done) => {
				if (!pass.write({
						someobject: true
					})) {
					pass.once('drain', () => {
						setTimeout(done, 200);
					})
				} else {
					setTimeout(done, 200);
				}
			}, () => {
				return ++x < 23;
			}, (err) => {
				console.log(err);
				pass.emit("error", "this is an error");
				console.log("waiting 5 seconds");
				pass.end();
			}, 1000 * 5);
		});
	});
});