stream.js
2.49 KB
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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
/**
* A lightweight readable stream implemention that handles event dispatching.
* Objects that inherit from streams should call init in their constructors.
*/
export default class Stream {
constructor() {
this.init();
}
init() {
this.listeners = {};
}
/**
* Add a listener for a specified event type.
* @param type {string} the event name
* @param listener {function} the callback to be invoked when an event of
* the specified type occurs
*/
on(type, listener) {
if (!this.listeners[type]) {
this.listeners[type] = [];
}
this.listeners[type].push(listener);
}
/**
* Remove a listener for a specified event type.
* @param type {string} the event name
* @param listener {function} a function previously registered for this
* type of event through `on`
*/
off(type, listener) {
let index;
if (!this.listeners[type]) {
return false;
}
index = this.listeners[type].indexOf(listener);
this.listeners[type].splice(index, 1);
return index > -1;
}
/**
* Trigger an event of the specified type on this stream. Any additional
* arguments to this function are passed as parameters to event listeners.
* @param type {string} the event name
*/
trigger(type) {
let callbacks;
let i;
let length;
let args;
callbacks = this.listeners[type];
if (!callbacks) {
return;
}
// Slicing the arguments on every invocation of this method
// can add a significant amount of overhead. Avoid the
// intermediate object creation for the common case of a
// single callback argument
if (arguments.length === 2) {
length = callbacks.length;
for (i = 0; i < length; ++i) {
callbacks[i].call(this, arguments[1]);
}
} else {
args = Array.prototype.slice.call(arguments, 1);
length = callbacks.length;
for (i = 0; i < length; ++i) {
callbacks[i].apply(this, args);
}
}
}
/**
* Destroys the stream and cleans up.
*/
dispose() {
this.listeners = {};
}
/**
* Forwards all `data` events on this stream to the destination stream. The
* destination stream should provide a method `push` to receive the data
* events as they arrive.
* @param destination {stream} the stream that will receive all `data` events
* @see http://nodejs.org/api/stream.html#stream_readable_pipe_destination_options
*/
pipe(destination) {
this.on('data', function(data) {
destination.push(data);
});
}
}