-
Notifications
You must be signed in to change notification settings - Fork 0
/
simpleObservable.go
86 lines (78 loc) · 1.88 KB
/
simpleObservable.go
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
package rx
import (
"sync"
)
// SimpleObservable represents an object that can emit a set of values over a
// period of time.
type SimpleObservable struct {
baseObservable
}
// NewSimpleObservable creates new simple observable.
func NewSimpleObservable(source Observable) Observable {
observable := SimpleObservable{
baseObservable: baseObservable{
_source: source,
_input: make(chan interface{}),
register: make(chan chan<- interface{}),
unregister: make(chan chan<- interface{}),
outputs: make(map[chan<- interface{}]*outputMetadata),
mutex: &sync.RWMutex{},
},
}
go observable.run()
return observable
}
// Subscribe registers Observer handlers for notifications it will emit.
func (o SimpleObservable) Subscribe(handlers ...EventHandler) *Subscription {
return o.subscribe(true, handlers...)
}
// Pipe is used to stitch together functional operators into a chain.
func (o SimpleObservable) Pipe(operations ...OperatorFunction) Observable {
var observable Observable = o
for _, operation := range operations {
observable = operation(observable)
}
return observable
}
// ToPromise returns a awaitable channel which closes after receiving one
// notification.
func (o SimpleObservable) ToPromise() func() chan interface{} {
channel := make(chan interface{})
sub := o.Subscribe(
NextFunc(func(item interface{}) {
channel <- item
}),
ErrorFunc(func(err error) {
channel <- err
}),
CompleteFunc(func() {
}),
)
var closed bool
return func() chan interface{} {
output := make(chan interface{})
func() {
value, ok := <-channel
if ok {
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer func() {
recover()
}()
wg.Done()
output <- value
}()
wg.Wait()
close(channel)
sub.Unsubscribe()
}
}()
if !closed {
closed = true
return output
}
close(output)
return output
}
}