From 8b8c6d128499655f14b7904ac9ae1693ee26ca60 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adri=C3=A1n=20Orive?= Date: Thu, 3 May 2018 17:19:35 +0200 Subject: [PATCH] Make queues size-aware leaving the old factory construct for infinite queues and providing a new one for sized queues MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Adrián Orive --- errors.go | 5 +++++ queue.go | 13 ++++++++++++- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/errors.go b/errors.go index 56db7c2..4993d15 100644 --- a/errors.go +++ b/errors.go @@ -7,4 +7,9 @@ var ( // closed. If encountered, the error should be considered terminal and // retries will not be successful. ErrSinkClosed = fmt.Errorf("events: sink closed") + + // ErrQueueFull is returned if a write is issued to a queue that does not + // have enough space to store an additional event. If encountered, further + // replies may be successful if any of the queue elements was consumed. + ErrQueueFull = fmt.Errorf("events: queue full") ) diff --git a/queue.go b/queue.go index 4bb770a..dd02bca 100644 --- a/queue.go +++ b/queue.go @@ -13,16 +13,23 @@ import ( type Queue struct { dst Sink events *list.List + limit int cond *sync.Cond mu sync.Mutex closed bool } -// NewQueue returns a queue to the provided Sink dst. +// NewQueue returns an infinite queue to the provided Sink dst. func NewQueue(dst Sink) *Queue { + return NewSizedQueue(dst, 0) +} + +// NewSizedQueue returns a sized queue to the provided Sink dst. +func NewSizedQueue(dst Sink, limit int) *Queue { eq := Queue{ dst: dst, events: list.New(), + limit: limit, } eq.cond = sync.NewCond(&eq.mu) @@ -40,6 +47,10 @@ func (eq *Queue) Write(event Event) error { return ErrSinkClosed } + if eq.limit > 0 && eq.events.Len() >= eq.limit { + return ErrQueueFull + } + eq.events.PushBack(event) eq.cond.Signal() // signal waiters