-
Notifications
You must be signed in to change notification settings - Fork 0
/
chan_utils.go
52 lines (41 loc) · 854 Bytes
/
chan_utils.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
package go_commons
type BufferChan struct {
bufferSize int
buff []string
closed chan struct{}
}
func NewBufferChan(size int) *BufferChan {
return &BufferChan{
bufferSize: size,
buff: make([]string, 0, size),
closed: make(chan struct{}),
}
}
func (b *BufferChan) BufferString(in chan string, out chan string) {
defer close(out)
var inflight = in
for delivery := range in {
b.buff = append(b.buff, delivery)
for len(b.buff) > 0 {
select {
case <-b.closed:
// closed before drained, drop in-flight
return
case delivery, consuming := <-inflight:
if consuming {
b.buff = append(b.buff, delivery)
} else {
inflight = nil
}
case out <- b.buff[0]:
b.buff = b.buff[1:]
}
}
}
}
func (b *BufferChan) Len() int {
return len(b.buff)
}
func (b *BufferChan) Close() {
close(b.closed)
}