commitlog

Documentation

Index

• Constants
• Variables
• type CommitLog
  • func New(options ...OptionFunc) (*CommitLog, error)
  • func (c *CommitLog) Append(p []byte) (offset int64, err error)
  • func (c *CommitLog) Close() error
  • func (c *CommitLog) DeleteAll() error
  • func (c *CommitLog) NewReader(offset int64) (io.Reader, error)
  • func (c *CommitLog) NewestOffset() int64
  • func (c *CommitLog) OldestOffset() int64
  • func (c *CommitLog) Segments() []*Segment
• type Compactor
  • func NewNamespaceCompactor(clog *CommitLog) Compactor
• type Log
  • func NewLogFromEntry(le LogEntry) Log
  • func (l Log) PutOffset(offset int64)
• type LogEntry
  • func ReadEntry(r io.Reader) (uint64, LogEntry, error)
  • func (le LogEntry) ModeOpToByte() byte
• type Mode
  • func (m Mode) String() string
• type OptionFunc
  • func WithMaxSegmentBytes(max int64) OptionFunc
  • func WithPath(path string) OptionFunc
• type Reader
  • func (r *Reader) Read(p []byte) (int, error)
• type Segment
  • func NewSegment(path, format string, baseOffset int64, maxBytes int64) (*Segment, error)
  • func (s *Segment) Close() error
  • func (s *Segment) FindOffsetPosition(offset uint64) (int64, error)
  • func (s *Segment) IsFull() bool
  • func (s *Segment) ReadAt(p []byte, off int64) (n int, err error)
  • func (s *Segment) Write(p []byte) (n int, err error)

Constants

const (
	// LogNameFormat defines the filename structure for active segments.
	LogNameFormat = "%020d.log"
)

Variables

var (

	// ErrEmptyPath will be returned if the provided path is an empty string
	ErrEmptyPath = errors.New("path is empty")
	// ErrSegmentNotFound is returned with no segment is found given the provided offset
	ErrSegmentNotFound = errors.New("segment not found")
)

var (
	// ErrOffsetNotFound is returned when the requested offset is not in the segment.
	ErrOffsetNotFound = errors.New("offset not found")
)

Types

type CommitLog

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

CommitLog is how the rest of the system will interact with the underlying log segments to persist and read messages.

func New

func New(options ...OptionFunc) (*CommitLog, error)

New creates a new CommitLog for persisting and reading messages.

The caller can configure the CommitLog by passing configuration options to the func.

Example:

c, err := New(
  WithPath("path/to/dir"),
  WithMaxSegmentBytes(1024))

An error is also returned when some configuration option is invalid

func (*CommitLog) Append

func (c *CommitLog) Append(p []byte) (offset int64, err error)

Append will convert the set the offset for the provided []byte and then persist it to the active segment.

func (*CommitLog) Close

func (c *CommitLog) Close() error

Close iterates over all segments and calls its Close() func.

func (*CommitLog) DeleteAll

func (c *CommitLog) DeleteAll() error

DeleteAll cleans out all data in the path.

func (*CommitLog) NewReader

func (c *CommitLog) NewReader(offset int64) (io.Reader, error)

NewReader returns an io.Reader based on the provider offset.

func (*CommitLog) NewestOffset

func (c *CommitLog) NewestOffset() int64

NewestOffset obtains the NextOffset of the current segment in use.

func (*CommitLog) OldestOffset

func (c *CommitLog) OldestOffset() int64

OldestOffset obtains the BaseOffset from the oldest segment on disk.

func (*CommitLog) Segments

func (c *CommitLog) Segments() []*Segment

Segments provides access to the underlying segments stored on disk.

type Compactor

type Compactor interface {
	Compact(uint64, []*Segment)
}

Compactor defines the necessary functions for performing compaction of log segments.

func NewNamespaceCompactor

func NewNamespaceCompactor(clog *CommitLog) Compactor

NewNamespaceCompactor creates a new Compactor to be used for tracking messages based on the namespace.

type Log

type Log []byte

Log is a alias type for []byte.

func NewLogFromEntry

func NewLogFromEntry(le LogEntry) Log

NewLogFromEntry takes the LogEntry and builds the underlying []byte to be stored.

func (Log) PutOffset

func (l Log) PutOffset(offset int64)

PutOffset sets the provided offset for the given Log.

type LogEntry

type LogEntry struct {
	Key       []byte
	Value     []byte
	Timestamp uint64
	Mode      Mode
	Op        ops.Op
}

LogEntry represents the high level representation of the message portion of each entry in the commit log.

func ReadEntry

func ReadEntry(r io.Reader) (uint64, LogEntry, error)

ReadEntry takes an io.Reader and returns a LogEntry.

func (LogEntry) ModeOpToByte

func (le LogEntry) ModeOpToByte() byte

ModeOpToByte converts the Mode and Op values into a single byte by performing bitwise operations. Mode is stored in bits 0 - 1 Op is stored in bits 2 - 4 bits 5 - 7 are currently unused

type Mode

type Mode int

Mode is a representation of where a in the process a reader is with respect to a given namespace.

const (
	Copy Mode = iota
	Sync
	Complete
)

currently supported Modes are Copy, Sync, and Complete

func (Mode) String

func (m Mode) String() string

type OptionFunc

type OptionFunc func(*CommitLog) error

OptionFunc is a function that configures a CommitLog. It is used in New.

func WithMaxSegmentBytes

func WithMaxSegmentBytes(max int64) OptionFunc

WithMaxSegmentBytes defines the maximum limit a log segment can reach before needing to create a new one.

func WithPath

func WithPath(path string) OptionFunc

WithPath defines the directory where all data will be stored.

type Reader

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

Reader implements io.Reader for use with reading from the commit log.

func (*Reader) Read

func (r *Reader) Read(p []byte) (int, error)

type Segment

type Segment struct {
	BaseOffset int64
	NextOffset int64
	Position   int64

	sync.Mutex
	// contains filtered or unexported fields
}

Segment handles reading and writing to the underlying files on disk.

func NewSegment

func NewSegment(path, format string, baseOffset int64, maxBytes int64) (*Segment, error)

NewSegment creates a new instance of Segment with the provided parameters and initializes its NextOffset and Position should the file be non-empty.

func (*Segment) Close

func (s *Segment) Close() error

Close closes the read/write access to the underlying file.

func (*Segment) FindOffsetPosition

func (s *Segment) FindOffsetPosition(offset uint64) (int64, error)

FindOffsetPosition attempts to find the provided offset position in the Segment.

func (*Segment) IsFull

func (s *Segment) IsFull() bool

IsFull determines whether the current size of the segment is greater than or equal to the maxBytes configured.

func (*Segment) ReadAt

func (s *Segment) ReadAt(p []byte, off int64) (n int, err error)

ReadAt calls ReadAt on the underlying log.

func (*Segment) Write

func (s *Segment) Write(p []byte) (n int, err error)