Queues and message iterators

Queues serve as buffers for storing incoming messages and making them available for consumption by compute functions. Users of the graph processing system can either make use of the built-in in-memory queue (see the next section) or inject their application-specific queue implementation as long as it adheres to the Queue interface, whose definition is listed as follows:

type Queue interface {
    // Cleanly shutdown the queue.
    Close() error

    // Enqueue inserts a message to the end of the queue.
    Enqueue(msg Message) error

    // PendingMessages returns true if the queue contains any messages.
    PendingMessages() bool

    // Flush drops all pending messages from the queue.
    DiscardMessages() error

    // Messages returns an iterator for accessing the queued messages.
    Messages() Iterator
}

The methods on the Queue interface are pretty standard for any type of queue system. A call to PendingMessages reveals whether the queue is currently empty, while a call to DiscardMessages can be used to flush any stored messages. The Enqueue method can be used to append a new Message to the queue, while the Messages method returns an Iterator for accessing the list of already enqueued messages. Since iterator implementations are typically coupled to the underlying queue system, Iterator is also defined as an interface:

type Iterator interface {
    // Next advances the iterator so that the next message can be retrieved
    // via a call to Message(). If no more messages are available or an
    // error occurs, Next() returns false.
    Next() bool

    // Message returns the message currently pointed to by the iterator.
    Message() Message

    // Error returns the last error that the iterator encountered.
    Error() error
}

This interface follows exactly the same iterator pattern that you should be familiar with from the previous chapters. Calling Next advances the iterator and returns a Boolean value to indicate whether more messages are available. After a successful call to Next, the current message can be retrieved by calling Message.