upcqueue

Documentation

Overview

Package upcqueue implements an unbounded producer/consumer queue. An unbounded producer/consumer queue is a concurrent buffer supporting multiple concurrent producers and consumers, with timeouts. The queue can be closed from either end, by the producer and/or the consumer. When closed, the contents are discarded, and subsequent operations return an error.

Note: the main reason to use a producer/consumer queue instead of a channel is to allow the consumer to close the channel. This queue can be used for many-to-many communication with multiple producers and/or multiple consumers. Any of the producers and any of the consumers are allowed to close the queue.

Index

• Variables
• type T
  • func New() *T
  • func (q *T) Close()
  • func (q *T) Get(cancel <-chan struct{}) (interface{}, error)
  • func (q *T) IsClosed() bool
  • func (q *T) Put(item interface{}) error
  • func (q *T) Shutdown() []interface{}
  • func (q *T) TryGet() (interface{}, error)

Variables

var (
	ErrQueueIsClosed = errors.New("queue is closed")
)

Types

type T

type T struct {
	// contains filtered or unexported fields
}

T is an unbounded producer/consumer queue. It fulfills the same purpose as a Go channel, the main advantage is that the Put() operation does not panic, even after the queue is closed. The main disadvantage is that the T can't be used in a select operation.

func New

func New() *T

New returns a producer/consumer queue.

func (*T) Close

func (q *T) Close()

Close closes the queue, without discarding the contents. All Put* operations currently running may, or may not, add their values to the queue. All Put* operations that happen-after the Close will fail.

func (*T) Get

func (q *T) Get(cancel <-chan struct{}) (interface{}, error)

Get returns the next item from the queue, or an error if the queue is closed or the operation is cancelled.

func (*T) IsClosed

func (q *T) IsClosed() bool

IsClosed returns whether the queue has been closed (with Close or Shutdown).

func (*T) Put

func (q *T) Put(item interface{}) error

Put adds an item to the queue, or returns an error if the queue is closed.

func (*T) Shutdown

func (q *T) Shutdown() []interface{}

Shutdown closes the queue and returns the contents. Any concurrent Get and Put operations might exchange values, but all operations that happen-after the Shutdown will fail.

func (*T) TryGet

func (q *T) TryGet() (interface{}, error)

TryGet returns the next item from the queue if there is one.