single

Documentation

Overview

Package single contains implementations of various buffer types where the buffer can only be consumed by a single thread (but any number of writers).

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 MmapBuffer
  • func NewMmapBuffer(config MmapBufferConfig, log log.Modular, stats metrics.Type) (*MmapBuffer, error)
  • func (f *MmapBuffer) Close()
  • func (f *MmapBuffer) CloseOnceEmpty()
  • func (f *MmapBuffer) NextMessage() (types.Message, error)
  • func (f *MmapBuffer) PushMessage(msg types.Message) (int, error)
  • func (f *MmapBuffer) ShiftMessage() (int, error)
• type MmapBufferConfig
  • func NewMmapBufferConfig() MmapBufferConfig
• type MmapCache
  • func NewMmapCache(config MmapCacheConfig, log log.Modular, stats metrics.Type) (*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

Variables

var (
	// ErrWrongTrackerLength means the length of a read tracker was not correct.
	ErrWrongTrackerLength = errors.New("tracker was unexpected length")

	// ErrNotEnoughSpace means the target disk lacked the space needed for a new
	// file.
	ErrNotEnoughSpace = errors.New("target disk is at capacity")
)

Types

type CachedMmap

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

CachedMmap is a struct containing a cached Mmap file and the file handler.

type Memory

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

Memory is 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 is config values for a purely memory based ring buffer type.

func NewMemoryConfig

func NewMemoryConfig() MemoryConfig

NewMemoryConfig creates a new MemoryConfig with default values.

type MmapBuffer

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

MmapBuffer is a buffer implemented around rotated memory mapped files.

func NewMmapBuffer

func NewMmapBuffer(config MmapBufferConfig, log log.Modular, stats metrics.Type) (*MmapBuffer, error)

NewMmapBuffer creates a memory-map based buffer.

func (*MmapBuffer) Close

func (f *MmapBuffer) Close()

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

func (*MmapBuffer) CloseOnceEmpty

func (f *MmapBuffer) CloseOnceEmpty()

CloseOnceEmpty closes the mmap buffer once the backlog reaches 0.

func (*MmapBuffer) NextMessage

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

NextMessage reads the next message, blocks until there's something to read.

func (*MmapBuffer) PushMessage

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

PushMessage pushes a new message, returns the backlog count.

func (*MmapBuffer) ShiftMessage

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

ShiftMessage removes the last message. Returns the backlog count.

type MmapBufferConfig

type MmapBufferConfig MmapCacheConfig

MmapBufferConfig is config options for a memory-map based buffer reader.

func NewMmapBufferConfig

func NewMmapBufferConfig() MmapBufferConfig

NewMmapBufferConfig creates a MmapBufferConfig oject with default values.

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, log log.Modular, stats metrics.Type) (*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 checks 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"`
	ReservedDiskSpace uint64 `json:"reserved_disk_space" yaml:"reserved_disk_space"`
}

MmapCacheConfig is config options for the MmapCache type.

func NewMmapCacheConfig

func NewMmapCacheConfig() MmapCacheConfig

NewMmapCacheConfig creates a new MmapCacheConfig oject with default values.