forked from ghemawat/stream
-
Notifications
You must be signed in to change notification settings - Fork 0
/
firstlast.go
83 lines (77 loc) · 1.77 KB
/
firstlast.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
package stream
// First yields the first n items that it receives.
func First[T Streamable](n int) Filter[T] {
return FilterFunc[T](func(arg Arg[T]) error {
seen := 0
for s := range arg.In {
if seen >= n {
break
}
arg.Out <- s
seen++
}
return nil
})
}
// DropFirst yields all items except for the first n items that it receives.
func DropFirst[T string](n int) Filter[T] {
return FilterFunc[T](func(arg Arg[T]) error {
seen := 0
for s := range arg.In {
if seen >= n {
arg.Out <- s
}
seen++
}
return nil
})
}
// ring is a circular buffer.
type ring[T Streamable] struct {
buf []T
next, n int
}
func newRing[T Streamable](n int) *ring[T] { return &ring[T]{buf: make([]T, n)} }
func (r *ring[T]) empty() bool { return r.n == 0 }
func (r *ring[T]) full() bool { return r.n == len(r.buf) }
func (r *ring[T]) pushBack(s T) {
r.buf[r.next] = s
r.next = (r.next + 1) % len(r.buf)
if r.n < len(r.buf) {
r.n++
}
}
func (r *ring[T]) popFront() T {
// The addition of len(r.buf) is so that the dividend is
// non-negative and therefore remainder is non-negative.
first := (r.next - r.n + len(r.buf)) % len(r.buf)
s := r.buf[first]
r.n--
return s
}
// Last yields the last n items that it receives.
func Last[T Streamable](n int) Filter[T] {
return FilterFunc[T](func(arg Arg[T]) error {
r := newRing[T](n)
for s := range arg.In {
r.pushBack(s)
}
for !r.empty() {
arg.Out <- r.popFront()
}
return nil
})
}
// DropLast yields all items except for the last n items that it receives.
func DropLast[T Streamable](n int) Filter[T] {
return FilterFunc[T](func(arg Arg[T]) error {
r := newRing[T](n)
for s := range arg.In {
if r.full() {
arg.Out <- r.popFront()
}
r.pushBack(s)
}
return nil
})
}