Merge pull request #94 from tylertreat-wf/rb_offer

Add Offer method to RingBuffer

Author: dustinhiatt-wf

queue/ring.go

@@ -72,13 +72,25 @@ func (rb *RingBuffer) init(size uint64) {
 // call will block until an item is added to the queue or Dispose is called
 // on the queue.  An error will be returned if the queue is disposed.
 func (rb *RingBuffer) Put(item interface{}) error {
+	_, err := rb.put(item, false)
+	return err
+}
+
+// Offer adds the provided item to the queue if there is space.  If the queue
+// is full, this call will return false.  An error will be returned if the
+// queue is disposed.
+func (rb *RingBuffer) Offer(item interface{}) (bool, error) {
+	return rb.put(item, true)
+}
+
+func (rb *RingBuffer) put(item interface{}, offer bool) (bool, error) {
 	var n *node
 	pos := atomic.LoadUint64(&rb.queue)
 	i := 0
 L:
 	for {
 		if atomic.LoadUint64(&rb.disposed) == 1 {
-			return ErrDisposed
+			return false, ErrDisposed
 		}
 
 		n = rb.nodes[pos&rb.mask]
@@ -94,6 +106,10 @@ L:
 			pos = atomic.LoadUint64(&rb.queue)
 		}
 
+		if offer {
+			return false, nil
+		}
+
 		if i == 10000 {
 			runtime.Gosched() // free up the cpu before the next iteration
 			i = 0
@@ -104,7 +120,7 @@ L:
 
 	n.data = item
 	atomic.StoreUint64(&n.position, pos+1)
-	return nil
+	return true, nil
 }
 
 // Get will return the next item in the queue.  This call will block

queue/ring_test.go

@@ -128,6 +128,27 @@ func TestPutToFull(t *testing.T) {
 	wg.Wait()
 }
 
+func TestOffer(t *testing.T) {
+	rb := NewRingBuffer(2)
+
+	ok, err := rb.Offer("foo")
+	assert.True(t, ok)
+	assert.Nil(t, err)
+	ok, err = rb.Offer("bar")
+	assert.True(t, ok)
+	assert.Nil(t, err)
+	ok, err = rb.Offer("baz")
+	assert.False(t, ok)
+	assert.Nil(t, err)
+
+	item, err := rb.Get()
+	assert.Nil(t, err)
+	assert.Equal(t, "foo", item)
+	item, err = rb.Get()
+	assert.Nil(t, err)
+	assert.Equal(t, "bar", item)
+}
+
 func TestRingGetEmpty(t *testing.T) {
 	rb := NewRingBuffer(3)