wal

Documentation

Index

• Constants
• Variables
• func NewPair(path string, opts WriteOptions) (*PairedReader, *PairedWriter, error)
• type PairedReader
  • func (r *PairedReader) BlockingNext() error
  • func (r *PairedReader) Next() bool
• type PairedWriter
  • func (r *PairedWriter) Write(d []byte) error
• type Position
  • func (p *Position) None() bool
• type SegmentReader
  • func NewSegmentReader(path string) (*SegmentReader, error)
  • func (r *SegmentReader) CRC() uint32
  • func (r *SegmentReader) Close() error
  • func (r *SegmentReader) Error() error
  • func (r *SegmentReader) Next() bool
  • func (s *SegmentReader) Pos() int64
  • func (r *SegmentReader) Seek(pos int64) error
  • func (r *SegmentReader) SeekTag(tag []byte) (int64, error)
  • func (r *SegmentReader) Value() []byte
• type SegmentWriter
  • func NewSegmentWriter(path string) (*SegmentWriter, error)
  • func (s *SegmentWriter) Clean() bool
  • func (s *SegmentWriter) Close() error
  • func (s *SegmentWriter) Pos() int64
  • func (s *SegmentWriter) SetSyncRate(dur time.Duration)
  • func (s *SegmentWriter) Size() int64
  • func (s *SegmentWriter) Truncate(pos int64) error
  • func (s *SegmentWriter) Write(data []byte) (int, error)
  • func (s *SegmentWriter) WriteTag(data []byte) error
• type WALReader
  • func BeginRecovery(path string, tag []byte) (*WALReader, error)
  • func NewReader(root string) (*WALReader, error)
  • func (r *WALReader) Close() error
  • func (r *WALReader) Error() error
  • func (r *WALReader) Next() bool
  • func (wal *WALReader) Pos() (Position, error)
  • func (wal *WALReader) Reset() error
  • func (wal *WALReader) Seek(p Position) error
  • func (wal *WALReader) SeekTag(tag []byte) (Position, error)
  • func (r *WALReader) Value() []byte
• type WALWriter
  • func New(root string) (*WALWriter, error)
  • func NewWithOptions(root string, opts WriteOptions) (*WALWriter, error)
  • func (wal *WALWriter) Close() error
  • func (wal *WALWriter) Pos() (Position, error)
  • func (wal *WALWriter) Write(data []byte) error
  • func (wal *WALWriter) WriteTag(tag []byte) error
• type WriteOptions
  • func (wo *WriteOptions) CalculateFromTotal(total int64)

Constants

const MaxSegmentSize = 16 * (1024 * 1024)

Variables

var DefaultWriteOptions = WriteOptions{
	SegmentSize: MaxSegmentSize,
	MaxSegments: 10,
}

Defaults to using 160MB of disk

var ErrCorruptCRC = errors.New("corrupt data detected")

var ErrNoData = errors.New("no data available")

var ErrNoSegments = errors.New("no segments")

Functions

func NewPair

func NewPair(path string, opts WriteOptions) (*PairedReader, *PairedWriter, error)

Types

type PairedReader

type PairedReader struct {
	*WALReader
	// contains filtered or unexported fields
}

func (*PairedReader) BlockingNext

func (r *PairedReader) BlockingNext() error

func (*PairedReader) Next

func (r *PairedReader) Next() bool

Next wraps the underlying WALReader's Next() method with a mutex which is shared by the writer. Since we are dealing with file IO, a simple mutex makes it safe for a paired reader and writer to run concurrently.

type PairedWriter

type PairedWriter struct {
	*WALWriter
	// contains filtered or unexported fields
}

func (*PairedWriter) Write

func (r *PairedWriter) Write(d []byte) error

type Position

type Position struct {
	Segment int   `json:"segment"`
	Offset  int64 `json:"offset"`
}

func (*Position) None

func (p *Position) None() bool

type SegmentReader

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

func NewSegmentReader

func NewSegmentReader(path string) (*SegmentReader, error)

func (*SegmentReader) CRC

func (r *SegmentReader) CRC() uint32

func (*SegmentReader) Close

func (r *SegmentReader) Close() error

func (*SegmentReader) Error

func (r *SegmentReader) Error() error

func (*SegmentReader) Next

func (r *SegmentReader) Next() bool

func (*SegmentReader) Pos

func (s *SegmentReader) Pos() int64

func (*SegmentReader) Seek

func (r *SegmentReader) Seek(pos int64) error

func (*SegmentReader) SeekTag

func (r *SegmentReader) SeekTag(tag []byte) (int64, error)

func (*SegmentReader) Value

func (r *SegmentReader) Value() []byte

type SegmentWriter

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

func NewSegmentWriter

func NewSegmentWriter(path string) (*SegmentWriter, error)

func (*SegmentWriter) Clean

func (s *SegmentWriter) Clean() bool

func (*SegmentWriter) Close

func (s *SegmentWriter) Close() error

func (*SegmentWriter) Pos

func (s *SegmentWriter) Pos() int64

func (*SegmentWriter) SetSyncRate

func (s *SegmentWriter) SetSyncRate(dur time.Duration)

func (*SegmentWriter) Size

func (s *SegmentWriter) Size() int64

func (*SegmentWriter) Truncate

func (s *SegmentWriter) Truncate(pos int64) error

func (*SegmentWriter) Write

func (s *SegmentWriter) Write(data []byte) (int, error)

func (*SegmentWriter) WriteTag

func (s *SegmentWriter) WriteTag(data []byte) error

type WALReader

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

func BeginRecovery

func BeginRecovery(path string, tag []byte) (*WALReader, error)

func NewReader

func NewReader(root string) (*WALReader, error)

func (*WALReader) Close

func (r *WALReader) Close() error

func (*WALReader) Error

func (r *WALReader) Error() error

func (*WALReader) Next

func (r *WALReader) Next() bool

func (*WALReader) Pos

func (wal *WALReader) Pos() (Position, error)

func (*WALReader) Reset

func (wal *WALReader) Reset() error

func (*WALReader) Seek

func (wal *WALReader) Seek(p Position) error

func (*WALReader) SeekTag

func (wal *WALReader) SeekTag(tag []byte) (Position, error)

func (*WALReader) Value

func (r *WALReader) Value() []byte

type WALWriter

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

func New

func New(root string) (*WALWriter, error)

func NewWithOptions

func NewWithOptions(root string, opts WriteOptions) (*WALWriter, error)

func (*WALWriter) Close

func (wal *WALWriter) Close() error

func (*WALWriter) Pos

func (wal *WALWriter) Pos() (Position, error)

func (*WALWriter) Write

func (wal *WALWriter) Write(data []byte) error

func (*WALWriter) WriteTag

func (wal *WALWriter) WriteTag(tag []byte) error

type WriteOptions

type WriteOptions struct {
	// The maximum size in bytes of each segment. When it reaches near this size,
	// a new segment will be created.
	SegmentSize int64

	// The maximum number of segments to keep on disk.
	MaxSegments int

	// If 0, sync is done after every write. Otherwise this controls
	// how often the WAL is sync'd to disk. Setting this can speed
	// up the WAL by sacrifing safety.
	SyncRate time.Duration
}

func (*WriteOptions) CalculateFromTotal

func (wo *WriteOptions) CalculateFromTotal(total int64)

Calculate the WriteOptions based on how much disk space the WAL should consume in total. The true on disk size might be more slightly more than this because the value is calculate against MaxSegmentSize, which is 16MB. If you wish to use a larger segment size (or more accurate one), then set SegmentSize and MaxSegments directly.