Code coverage report for lib\streams\ResponseStream.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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
1
1
 
 
1
4
3
3
3
3
2
 
1
1
 
1
 
 
1
1
1
1
1
1
 
1
 
 
 
1
1
1
1
1
 
 
 
1
1
1
 
 
1
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
 
 
 
 
 
 
 
 
1
 
 
 
 
 
1
 
 
 
 
 
1
'use strict';
var kefir = require('kefir');
var BaseStream = require('./BaseStream.js');
 
 
var ResponseStream = function ResponseStream(from) {
    if (this instanceof ResponseStream) {
        this.init();
        this.finsishEmitter = kefir.emitter();
        this.closeEmitter = kefir.emitter();
        if (from === undefined) {
            return this;
        }
        this.from(from);
        return this;
    }
    return new ResponseStream(from);
};
 
ResponseStream.prototype = new BaseStream();
ResponseStream.prototype.constructor = ResponseStream;
ResponseStream.prototype.queue = [];
ResponseStream.prototype.preQueue = [];
ResponseStream.prototype.closeEmitter = null;
ResponseStream.prototype.finsishEmitter = null;
 
ResponseStream.prototype.use = function processing(fn) {
    this.queue.push(fn);
};
 
ResponseStream.prototype.from = function from(stream) {
    var self = this;
    self.broke();
    self.origin = stream;
    self.pipeListener = function requestEmitter(request) {
        request = self.preprocessing(x);
        self.emitter.emit(request);
    }
    stream.onValue(self.pipeListener);
    stream.onEnd(self.broke);
    return this;
}
 
ResponseStream.prototype.process = function process(x, queue) {
    var queue = queue || this.queue;
    var data = x;
    for (var i in queue) {
        data = queue[i](x);
    }
    return data;
};
 
 
ResponseStream.prototype.preprocessing = function preprocessing(x) {
    var self = this;
    var preQueue = self.preQueue;
    x.res.on('close', function() {
        self.closeEmitter.emit(x);
    });
    x.res.on('finish', function() {
        self.finsishEmitter.emit(x);
    });
    x = self.process(x, preQueue);
    return x;
};
 
//Emits on all responses 'close' event
ResponseStream.prototype.onClose = function onClose(fn) {
    self.closeEmitter.onValue(fn);
    return self;
};
 
//Emits on all responses 'finish' event
ResponseStream.prototype.onFinish = function onFinish(fn) {
    self.finsishEmitter.onValue(fn);
    return self;
};
 
 
module.exports = ResponseStream;