parallel

Documentation

Overview

Package parallel contains implementations of various buffer types where the buffer can be consumed by any number of parallel consumer threads. Therefore, since it is possible for consumers to requeue a message if the propagation failed, it is possible for messages to be consumed out of sequence.

Index

• type AckFunc
• type Memory
  • func NewMemory(capacity int) *Memory
  • func (m *Memory) Close()
  • func (m *Memory) CloseOnceEmpty()
  • func (m *Memory) NextMessage() (types.Message, AckFunc, error)
  • func (m *Memory) PushMessage(msg types.Message) (int, error)

Types

type AckFunc

type AckFunc func(ack bool) (int, error)

AckFunc is a func returned when a message is read from a parallel buffer. The func should be called when the message is finished with, and a flag indicates whether the message was successfully propagated and can be removed from the buffer. Returns the current backlog of the buffer in bytes, or an error if the message was not successfully removed.

If an error is returned it is safe to call the function again. Otherwise, it is not.

It is safe to call this func even if the buffer has closed.

type Memory

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

Memory is a parallel buffer implementation that allows multiple parallel consumers to read and purge messages from the buffer asynchronously.

func NewMemory

func NewMemory(capacity int) *Memory

NewMemory creates a memory based parallel buffer.

func (*Memory) Close

func (m *Memory) Close()

Close closes the Buffer so that blocked readers or writers become unblocked.

func (*Memory) CloseOnceEmpty

func (m *Memory) CloseOnceEmpty()

CloseOnceEmpty closes the Buffer once the buffer has been emptied, this is a way for a writer to signal to a reader that it is finished writing messages, and therefore the reader can close once it is caught up. This call blocks until the close is completed.

func (*Memory) NextMessage

func (m *Memory) NextMessage() (types.Message, AckFunc, error)

NextMessage reads the next oldest message, the message is preserved until the returned AckFunc is called.

func (*Memory) PushMessage

func (m *Memory) PushMessage(msg types.Message) (int, error)

PushMessage adds a new message to the stack. Returns the backlog in bytes.