初始化

康凯

2022-05-20 aa4c235a8ca67ea8b731f90c951a465e92c0a865

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
'use strict';
const Observable = require('any-observable');
 
function or(option, alternate, required) {
    const result = option === false ? false : option || alternate;
 
    if ((required && !result) || (result && typeof result !== 'string')) {
        throw new TypeError(alternate + 'Event must be a string.');
    }
 
    return result;
}
 
module.exports = (stream, opts) => {
    opts = opts || {};
 
    let complete = false;
    let dataListeners = [];
    const awaited = opts.await;
    const dataEvent = or(opts.dataEvent, 'data', true);
    const errorEvent = or(opts.errorEvent, 'error');
    const endEvent = or(opts.endEvent, 'end');
 
    function cleanup() {
        complete = true;
        dataListeners.forEach(listener => {
            stream.removeListener(dataEvent, listener);
        });
        dataListeners = null;
    }
 
    const completion = new Promise((resolve, reject) => {
        function onEnd(result) {
            if (awaited) {
                awaited.then(resolve);
            } else {
                resolve(result);
            }
        }
 
        if (endEvent) {
            stream.once(endEvent, onEnd);
        } else if (awaited) {
            onEnd();
        }
 
        if (errorEvent) {
            stream.once(errorEvent, reject);
        }
 
        if (awaited) {
            awaited.catch(reject);
        }
    }).catch(err => {
        cleanup();
        throw err;
    }).then(result => {
        cleanup();
        return result;
    });
 
    return new Observable(observer => {
        completion
            .then(observer.complete.bind(observer))
            .catch(observer.error.bind(observer));
 
        if (complete) {
            return null;
        }
 
        const onData = data => {
            observer.next(data);
        };
 
        stream.on(dataEvent, onData);
        dataListeners.push(onData);
 
        return () => {
            stream.removeListener(dataEvent, onData);
 
            if (complete) {
                return;
            }
 
            const idx = dataListeners.indexOf(onData);
 
            if (idx !== -1) {
                dataListeners.splice(idx, 1);
            }
        };
    });
};