-
Notifications
You must be signed in to change notification settings - Fork 1
/
channelqueue.go
208 lines (186 loc) · 4.97 KB
/
channelqueue.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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
package channelqueue
import "github.com/gammazero/deque"
// ChannelQueue uses a queue to buffer data between input and output channels.
type ChannelQueue[T any] struct {
input, output chan T
length chan int
capacity int
}
// New creates a new ChannelQueue with the specified buffer capacity.
//
// A capacity < 0 specifies unlimited capacity. Unbuffered behavior is not
// supported; use a normal channel for that. Use caution if specifying an
// unlimited capacity since storage is still limited by system resources.
func New[T any](capacity int) *ChannelQueue[T] {
if capacity == 0 {
panic("unbuffered behavior not supported")
}
if capacity < 0 {
capacity = -1
}
cq := &ChannelQueue[T]{
input: make(chan T),
output: make(chan T),
length: make(chan int),
capacity: capacity,
}
go cq.bufferData()
return cq
}
// NewRing creates a new ChannelQueue with the specified buffer capacity, and
// circular buffer behavior. When the buffer is full, writing an additional
// item discards the oldest buffered item.
func NewRing[T any](capacity int) *ChannelQueue[T] {
if capacity < 1 {
return New[T](capacity)
}
cq := &ChannelQueue[T]{
input: make(chan T),
output: make(chan T),
length: make(chan int),
capacity: capacity,
}
if capacity == 1 {
go cq.oneBufferData()
} else {
go cq.ringBufferData()
}
return cq
}
// In returns the write side of the channel.
func (cq *ChannelQueue[T]) In() chan<- T {
return cq.input
}
// Out returns the read side of the channel.
func (cq *ChannelQueue[T]) Out() <-chan T {
return cq.output
}
// Len returns the number of items buffered in the channel.
func (cq *ChannelQueue[T]) Len() int {
return <-cq.length
}
// Cap returns the capacity of the channel.
func (cq *ChannelQueue[T]) Cap() int {
return cq.capacity
}
// Close closes the input channel. Additional input will panic, output will
// continue to be readable until there is no more data, and then the output
// channel is closed.
func (cq *ChannelQueue[T]) Close() {
close(cq.input)
}
// bufferData is the goroutine that transfers data from the In() chan to the
// buffer and from the buffer to the Out() chan.
func (cq *ChannelQueue[T]) bufferData() {
var buffer deque.Deque[T]
var output chan T
var next, zero T
inputChan := cq.input
input := inputChan
for input != nil || output != nil {
select {
case elem, open := <-input:
if open {
// Push data from input chan to buffer.
buffer.PushBack(elem)
} else {
// Input chan closed; do not select input chan.
input = nil
inputChan = nil
}
case output <- next:
// Wrote buffered data to output chan. Remove item from buffer.
buffer.PopFront()
case cq.length <- buffer.Len():
}
if buffer.Len() == 0 {
// No buffered data; do not select output chan.
output = nil
next = zero // set to zero to GC value
} else {
// Try to write it to output chan.
output = cq.output
next = buffer.Front()
}
if cq.capacity != -1 {
// If buffer at capacity, then stop accepting input.
if buffer.Len() >= cq.capacity {
input = nil
} else {
input = inputChan
}
}
}
close(cq.output)
close(cq.length)
}
// ringBufferData is the goroutine that transfers data from the In() chan to
// the buffer and from the buffer to the Out() chan, with circular buffer
// behavior of discarding the oldest item when writing to a full buffer.
func (cq *ChannelQueue[T]) ringBufferData() {
var buffer deque.Deque[T]
var output chan T
var next, zero T
input := cq.input
for input != nil || output != nil {
select {
case elem, open := <-input:
if open {
// Push data from input chan to buffer.
buffer.PushBack(elem)
if buffer.Len() > cq.capacity {
buffer.PopFront()
}
} else {
// Input chan closed; do not select input chan.
input = nil
}
case output <- next:
// Wrote buffered data to output chan. Remove item from buffer.
buffer.PopFront()
case cq.length <- buffer.Len():
}
if buffer.Len() == 0 {
// No buffered data; do not select output chan.
output = nil
next = zero // set to zero to GC value
} else {
// Try to write it to output chan.
output = cq.output
next = buffer.Front()
}
}
close(cq.output)
close(cq.length)
}
// oneBufferData is the same as ringBufferData, but with a buffer size of 1.
func (cq *ChannelQueue[T]) oneBufferData() {
var bufLen int
var output chan T
var next, zero T
input := cq.input
for input != nil || output != nil {
select {
case elem, open := <-input:
if open {
// Push data from input chan to buffer.
next = elem
bufLen = 1
// Try to write it to output chan.
output = cq.output
} else {
// Input chan closed; do not select input chan.
input = nil
}
case output <- next:
// Wrote buffered data to output chan. Remove item from buffer.
bufLen = 0
next = zero // set to zero to GC value
// No buffered data; do not select output chan.
output = nil
case cq.length <- bufLen:
}
}
close(cq.output)
close(cq.length)
}