/usr/share/gocode/src/gopkg.in/eapache/channels.v1/infinite_channel.go is in golang-gopkg-eapache-channels.v1-dev 1.1.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | package channels
import "github.com/eapache/queue"
// InfiniteChannel implements the Channel interface with an infinite buffer between the input and the output.
type InfiniteChannel struct {
input, output chan interface{}
length chan int
buffer *queue.Queue
}
func NewInfiniteChannel() *InfiniteChannel {
ch := &InfiniteChannel{
input: make(chan interface{}),
output: make(chan interface{}),
length: make(chan int),
buffer: queue.New(),
}
go ch.infiniteBuffer()
return ch
}
func (ch *InfiniteChannel) In() chan<- interface{} {
return ch.input
}
func (ch *InfiniteChannel) Out() <-chan interface{} {
return ch.output
}
func (ch *InfiniteChannel) Len() int {
return <-ch.length
}
func (ch *InfiniteChannel) Cap() BufferCap {
return Infinity
}
func (ch *InfiniteChannel) Close() {
close(ch.input)
}
func (ch *InfiniteChannel) infiniteBuffer() {
var input, output chan interface{}
var next interface{}
input = ch.input
for input != nil || output != nil {
select {
case elem, open := <-input:
if open {
ch.buffer.Add(elem)
} else {
input = nil
}
case output <- next:
ch.buffer.Remove()
case ch.length <- ch.buffer.Length():
}
if ch.buffer.Length() > 0 {
output = ch.output
next = ch.buffer.Peek()
} else {
output = nil
next = nil
}
}
close(ch.output)
close(ch.length)
}
|