tools/readable.mjs at master · antcolag/tools · GitHub

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
/**
 * Object extension tools.
 * @module
 */

 import {
	injectProperties
} from "./tools.mjs"

import {
	defer,
	Semaphore
} from "./utils.mjs"

import {
	noop
} from "./operation.mjs"

export const BUFFER = Symbol('buffer')
export const WAIT = Symbol('wait')

export const HANDLERS = {
	read,
	broadcast,
	detach
}

export default function readable(
		broadcast = HANDLERS.broadcast,
		read = HANDLERS.read
	) {
	var opt = {
		broadcast: broadcast,
		read: read
	}
	return injectProperties.call(this, opt)
}

function broadcast(){
	this[BUFFER] = arguments
	flush.apply(this, arguments)
	init.call(this)
}

function read(time){
	if(time < 0){
		return this[BUFFER]
	}
	if(!this[WAIT]){
		init.call(this)
	}
	return (time ? Semaphore.race([
		this[WAIT],
		defer(noop)
	]) : this[WAIT])
}

function init(){
	this[WAIT] = new Semaphore()
	this[WAIT].catch( e => {
		this[BUFFER] = e
	})
}

function flush(){
	this[WAIT]
	&& this[WAIT].resolve
	&& this[WAIT].resolve(this[BUFFER])
}

function detach() {
	this[WAIT].reject && this[WAIT].reject(this[BUFFER])
}