domain

Documentation

Index

• Constants
• Variables
• type Chunk
• type ChunkAllocator
  • func NewChunkAllocator(options ...ChunkAllocatorOption) *ChunkAllocator
  • func (a *ChunkAllocator) FreePercent() float64
  • func (a *ChunkAllocator) HasFree() bool
• type ChunkAllocatorOption
  • func WithAllocatorSize(size int) ChunkAllocatorOption
  • func WithFullChThreshold(threshold int) ChunkAllocatorOption
• type PollResult
• type Segment
  • func NewSegment(topic types.Topic, hashRange types.HashRange, from uuid.Time, ...) Segment
  • func (s *Segment) Append(key types.Key, data []byte) (id uuid.UUID, err error)
  • func (s *Segment) Close()
  • func (s *Segment) Commit(n int)
  • func (s *Segment) Get(key types.Key) (msgId uuid.UUID, data []byte, ok bool)
  • func (s *Segment) Poll(from time.Time, fromMsgId uuid.UUID) (res PollResult, err error)
  • func (s *Segment) String() string
  • func (s *Segment) WaitCommitted()
• type SegmentOption
  • func WithInitialKeysCapacityKV(capacity int) SegmentOption
  • func WithMaxKeysKV(maxKeysKV int) SegmentOption
  • func WithMaxSegmentChunks(maxSegmentChunks int) SegmentOption

Constants

const (
	DefaultAllocatorSize   = 100 * 1024 * 1024 // 100 MB
	DefaultFullChThreshold = 0
)

const (
	DefaultMaxKeysKV            = 10_000_000
	DefaultInitialKeyCapacityKV = 100_000
	DefaultMaxSegmentChunks     = 100
	DefaultMaxUncommitted       = 50_000
)

const ChunkSize = 9 * 1024 * 1024 // 9 MB

Variables

var (
	ErrSegmentFull    = errors.New("segment is full")
	ErrChunkFull      = errors.New("chunk is full")
	ErrTooLargeData   = errors.New("data is too large to be pushed")
	ErrOutOfHashRange = errors.New("hash is out of range")
)

var ErrOutOfChunks = errors.New("out of chunks")

Types

type Chunk

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

type ChunkAllocator

type ChunkAllocator struct {
	FullCh chan struct{}

	FullChThreshold int // Number of free chunks to trigger FullCh, should be high in production but can be low in dev and teesting.
	// contains filtered or unexported fields
}

func NewChunkAllocator

func NewChunkAllocator(options ...ChunkAllocatorOption) *ChunkAllocator

func (*ChunkAllocator) FreePercent

func (a *ChunkAllocator) FreePercent() float64

func (*ChunkAllocator) HasFree

func (a *ChunkAllocator) HasFree() bool

type ChunkAllocatorOption

type ChunkAllocatorOption func(*ChunkAllocator)

func WithAllocatorSize

func WithAllocatorSize(size int) ChunkAllocatorOption

func WithFullChThreshold

func WithFullChThreshold(threshold int) ChunkAllocatorOption

type PollResult

type PollResult struct {
	HashRange types.HashRange
	From      time.Time
	To        time.Time
	LastMsgId uuid.UUID
	Data      []byte
	EOF       bool
}

type Segment

type Segment struct {
	Topic            types.Topic
	HashRange        types.HashRange
	From             uuid.Time
	Finished         bool
	Closed           bool
	MaxKeysKV        int
	MaxSegmentChunks int
	// contains filtered or unexported fields
}

func NewSegment

func NewSegment(topic types.Topic, hashRange types.HashRange, from uuid.Time, allocator *ChunkAllocator, options ...SegmentOption) Segment

func (*Segment) Append

func (s *Segment) Append(key types.Key, data []byte) (id uuid.UUID, err error)

func (*Segment) Close

func (s *Segment) Close()

Close returns the chunks to the allocator.

func (*Segment) Commit

func (s *Segment) Commit(n int)

func (*Segment) Get

func (s *Segment) Get(key types.Key) (msgId uuid.UUID, data []byte, ok bool)

func (*Segment) Poll

func (s *Segment) Poll(from time.Time, fromMsgId uuid.UUID) (res PollResult, err error)

func (*Segment) String

func (s *Segment) String() string

func (*Segment) WaitCommitted

func (s *Segment) WaitCommitted()

Wait until everything is committed, should only be called on same routine that does Append.

type SegmentOption

type SegmentOption func(*Segment)

func WithInitialKeysCapacityKV

func WithInitialKeysCapacityKV(capacity int) SegmentOption

func WithMaxKeysKV

func WithMaxKeysKV(maxKeysKV int) SegmentOption

func WithMaxSegmentChunks

func WithMaxSegmentChunks(maxSegmentChunks int) SegmentOption