Document Ring

This commit is contained in:
Simon Ser 2020-02-17 16:09:35 +01:00
parent 528c402bd0
commit 70d811f5a8
No known key found for this signature in database
GPG Key ID: 0FDE7BE0E88F5E48
2 changed files with 22 additions and 2 deletions

View File

@ -287,7 +287,7 @@ func (dc *downstreamConn) register() error {
} }
} }
consumer, ch := uc.ring.Consumer(seqPtr) consumer, ch := uc.ring.NewConsumer(seqPtr)
go func() { go func() {
for { for {
var closed bool var closed bool

22
ring.go
View File

@ -17,6 +17,7 @@ type Ring struct {
consumers []*RingConsumer consumers []*RingConsumer
} }
// NewRing creates a new ring buffer.
func NewRing(capacity int) *Ring { func NewRing(capacity int) *Ring {
return &Ring{ return &Ring{
buffer: make([]*irc.Message, capacity), buffer: make([]*irc.Message, capacity),
@ -24,6 +25,7 @@ func NewRing(capacity int) *Ring {
} }
} }
// Produce appends a new message to the ring buffer.
func (r *Ring) Produce(msg *irc.Message) { func (r *Ring) Produce(msg *irc.Message) {
r.lock.Lock() r.lock.Lock()
defer r.lock.Unlock() defer r.lock.Unlock()
@ -42,7 +44,17 @@ func (r *Ring) Produce(msg *irc.Message) {
} }
} }
func (r *Ring) Consumer(seq *uint64) (*RingConsumer, <-chan struct{}) { // NewConsumer creates a new ring buffer consumer.
//
// If seq is nil, the consumer will get messages starting from the last
// producer message. If seq is non-nil, the consumer will get messages starting
// from the specified history sequence number (see RingConsumer.Close).
//
// The returned channel yields a value each time the consumer has a new message
// available. Consume should be called to drain the consumer.
//
// The consumer can only be used from a single goroutine.
func (r *Ring) NewConsumer(seq *uint64) (*RingConsumer, <-chan struct{}) {
consumer := &RingConsumer{ consumer := &RingConsumer{
ring: r, ring: r,
ch: make(chan struct{}, 1), ch: make(chan struct{}, 1),
@ -63,6 +75,7 @@ func (r *Ring) Consumer(seq *uint64) (*RingConsumer, <-chan struct{}) {
return consumer, consumer.ch return consumer, consumer.ch
} }
// RingConsumer is a ring buffer consumer.
type RingConsumer struct { type RingConsumer struct {
ring *Ring ring *Ring
cur uint64 cur uint64
@ -78,6 +91,8 @@ func (rc *RingConsumer) diff() uint64 {
return rc.ring.cur - rc.cur return rc.ring.cur - rc.cur
} }
// Peek returns the next pending message if any without consuming it. A nil
// message is returned if no message is available.
func (rc *RingConsumer) Peek() *irc.Message { func (rc *RingConsumer) Peek() *irc.Message {
if rc.closed { if rc.closed {
panic("jounce: RingConsumer.Peek called after Close") panic("jounce: RingConsumer.Peek called after Close")
@ -102,6 +117,8 @@ func (rc *RingConsumer) Peek() *irc.Message {
return msg return msg
} }
// Consume consumes and returns the next pending message. A nil message is
// returned if no message is available.
func (rc *RingConsumer) Consume() *irc.Message { func (rc *RingConsumer) Consume() *irc.Message {
msg := rc.Peek() msg := rc.Peek()
if msg != nil { if msg != nil {
@ -110,6 +127,9 @@ func (rc *RingConsumer) Consume() *irc.Message {
return msg return msg
} }
// Close stops consuming messages. The consumer channel will be closed. The
// current history sequence number is returned. It can be provided later as an
// argument to Ring.NewConsumer to resume the message stream.
func (rc *RingConsumer) Close() uint64 { func (rc *RingConsumer) Close() uint64 {
rc.ring.lock.Lock() rc.ring.lock.Lock()
for i := range rc.ring.consumers { for i := range rc.ring.consumers {