ring

Documentation

Overview

Package ring - Defines strategies for generating and managing ring buffers.

Index

• Variables
• type CachedMmap
• type Memory
  • func NewMemory(config MemoryConfig) *Memory
  • func (m *Memory) Close()
  • func (m *Memory) CloseOnceEmpty()
  • func (m *Memory) NextMessage() (types.Message, error)
  • func (m *Memory) PushMessage(msg types.Message) (int, error)
  • func (m *Memory) ShiftMessage() (int, error)
• type MemoryConfig
  • func NewMemoryConfig() MemoryConfig
• type MessageStack
• type Mmap
  • func NewMmap(config MmapConfig, log log.Modular, stats metrics.Type) (*Mmap, error)
  • func (f *Mmap) Close()
  • func (f *Mmap) CloseOnceEmpty()
  • func (f *Mmap) NextMessage() (types.Message, error)
  • func (f *Mmap) PushMessage(msg types.Message) (int, error)
  • func (f *Mmap) ShiftMessage() (int, error)
• type MmapCache
  • func NewMmapCache(config MmapCacheConfig) (*MmapCache, error)
  • func (f *MmapCache) Delete(index int) error
  • func (f *MmapCache) EnsureCached(index int) error
  • func (f *MmapCache) Get(index int) []byte
  • func (f *MmapCache) GetTracker() []byte
  • func (f *MmapCache) IsCached(index int) bool
  • func (f *MmapCache) Remove(index int) error
  • func (f *MmapCache) RemoveAll() error
• type MmapCacheConfig
  • func NewMmapCacheConfig() MmapCacheConfig
• type MmapConfig
  • func NewMmapConfig() MmapConfig

Variables

var ErrWrongTrackerLength = errors.New("tracker was unexpected length")

ErrWrongTrackerLength - The length of a read tracker was not correct.

Types

type CachedMmap

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

CachedMmap - A struct containing a cached Mmap file and the file handler.

type Memory

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

Memory - A purely memory based ring buffer. This buffer blocks when the buffer is full.

func NewMemory

func NewMemory(config MemoryConfig) *Memory

NewMemory - Creates a new memory based ring buffer.

func (*Memory) Close

func (m *Memory) Close()

Close - Unblocks any blocked calls and prevents further writing to the block.

func (*Memory) CloseOnceEmpty

func (m *Memory) CloseOnceEmpty()

CloseOnceEmpty - Closes the memory buffer once the backlog reaches 0.

func (*Memory) NextMessage

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

NextMessage - Reads the next message, this call blocks until there's something to read.

func (*Memory) PushMessage

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

PushMessage - Pushes a new message onto the block, returns the backlog count.

func (*Memory) ShiftMessage

func (m *Memory) ShiftMessage() (int, error)

ShiftMessage - Removes the last message from the block. Returns the backlog count.

type MemoryConfig

type MemoryConfig struct {
	Limit int `json:"limit" yaml:"limit"`
}

MemoryConfig - Config values for a purely memory based ring buffer type.

func NewMemoryConfig

func NewMemoryConfig() MemoryConfig

NewMemoryConfig - Create a new MemoryConfig with default values.

type MessageStack

type MessageStack interface {
	// ShiftMessage - Remove the oldest message from the stack. Returns the backlog in bytes.
	ShiftMessage() (int, error)

	// NextMessage - Read the oldest message, the message is preserved until ShiftMessage is called.
	NextMessage() (types.Message, error)

	// PushMessage - Add a new message to the stack. Returns the backlog in bytes.
	PushMessage(types.Message) (int, error)

	// CloseOnceEmpty - Close the MessageStack 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.
	CloseOnceEmpty()

	// Close - Close the MessageStack so that blocked readers or writers become unblocked.
	Close()
}

MessageStack - Represents a method of stacking messages for persistence or fast buffering.

type Mmap

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

Mmap - A ring buffer implemented around memory mapped files. NOTE: Currently the 'ring' is non existent since files are created as memory is needed. However, we may in future want to limit the space used on disk and therefore looping will need to be implemented.

func NewMmap

func NewMmap(config MmapConfig, log log.Modular, stats metrics.Type) (*Mmap, error)

NewMmap - Creates a block for buffering serialized messages.

func (*Mmap) Close

func (f *Mmap) Close()

Close - Unblocks any blocked calls and prevents further writing to the block.

func (*Mmap) CloseOnceEmpty

func (f *Mmap) CloseOnceEmpty()

CloseOnceEmpty - Closes the mmap buffer once the backlog reaches 0.

func (*Mmap) NextMessage

func (f *Mmap) NextMessage() (types.Message, error)

NextMessage - Reads the next message, this call blocks until there's something to read.

func (*Mmap) PushMessage

func (f *Mmap) PushMessage(msg types.Message) (int, error)

PushMessage - Pushes a new message onto the block, returns the backlog count.

func (*Mmap) ShiftMessage

func (f *Mmap) ShiftMessage() (int, error)

ShiftMessage - Removes the last message from the block. Returns the backlog count.

type MmapCache

type MmapCache struct {
	*sync.Cond
	// contains filtered or unexported fields
}

MmapCache - Keeps track of any Mmap files cached in memory and cleans up resources as they are unclaimed. This type works similarly to sync.Cond, where if you wish to use it you need to lock it.

func NewMmapCache

func NewMmapCache(config MmapCacheConfig) (*MmapCache, error)

NewMmapCache - Creates a cache for managing open mmap files.

func (*MmapCache) Delete

func (f *MmapCache) Delete(index int) error

Delete - Deletes the file for an index.

func (*MmapCache) EnsureCached

func (f *MmapCache) EnsureCached(index int) error

EnsureCached - Check that a particular index is cached, and if not then read the index, this call blocks until either the index is successfully cached or an error occurs.

func (*MmapCache) Get

func (f *MmapCache) Get(index int) []byte

Get - Returns the []byte from a memory mapped file index.

func (*MmapCache) GetTracker

func (f *MmapCache) GetTracker() []byte

GetTracker - Returns the []byte from the tracker file memory mapping.

func (*MmapCache) IsCached

func (f *MmapCache) IsCached(index int) bool

IsCached - Returns a bool indicating whether the current memory mapped file index is cached.

func (*MmapCache) Remove

func (f *MmapCache) Remove(index int) error

Remove - Removes the index from our cache, the file is NOT deleted.

func (*MmapCache) RemoveAll

func (f *MmapCache) RemoveAll() error

RemoveAll - Removes all indexes from the cache as well as the tracker.

type MmapCacheConfig

type MmapCacheConfig struct {
	Path          string `json:"directory" yaml:"directory"`
	FileSize      int    `json:"file_size" yaml:"file_size"`
	RetryPeriodMS int    `json:"retry_period_ms" yaml:"retry_period_ms"`
	CleanUp       bool   `json:"clean_up" yaml:"clean_up"`
}

MmapCacheConfig - Config options for the MmapCache type.

func NewMmapCacheConfig

func NewMmapCacheConfig() MmapCacheConfig

NewMmapCacheConfig - Creates a new MmapCacheConfig oject with default values.

type MmapConfig

type MmapConfig MmapCacheConfig

MmapConfig - Config options for a memory-map based ring buffer.

func NewMmapConfig

func NewMmapConfig() MmapConfig

NewMmapConfig - Creates a new MmapConfig oject with default values.