message

Documentation

Index

• Constants
• Variables
• func MinOffset(offsets map[int64]struct{}) int64
• func Size(m Message, v Version) int64
• type Message
  • func Gen(count int) []Message
• type Reader
  • func OpenReader(path string, offset int64) (r *Reader, retErr error)
  • func OpenReaderMem(path string, offset int64) (r *Reader, retErr error)
  • func (r *Reader) Close() error
  • func (r *Reader) Consume(position, maxPosition int64, maxCount int64) ([]Message, error)
  • func (r *Reader) Get(position int64) (msg Message, err error)
  • func (r *Reader) InitialPosition() int64
  • func (r *Reader) Read(position int64) (msg Message, nextPosition int64, err error)
  • func (r *Reader) Version() Version
• type Version
  • func (v Version) String() string
• type Writer
  • func OpenWriter(path string, offset int64, newVersion Version) (w *Writer, retErr error)
  • func (w *Writer) Close() error
  • func (w *Writer) Size() int64
  • func (w *Writer) Sync() error
  • func (w *Writer) SyncAndClose() error
  • func (w *Writer) Version() Version
  • func (w *Writer) Write(m Message) (int64, error)

Constants

const (
	// OffsetOldest represents the smallest offset still available
	// Use it to consume all messages, starting at the first available
	OffsetOldest int64 = -2
	// OffsetNewest represents the offset that will be used for the next produce
	// Use it to consume only new messages
	OffsetNewest int64 = -1
	// OffsetInvalid is the offset returned when error is detected
	OffsetInvalid int64 = -3
)

const HeaderSize = int64(len(magic) + 2) // magic + Version byte + reserved byte

Variables

var (
	VUnknown         = Version{}
	V1               = Version{/* contains filtered or unexported fields */}
	V2               = Version{/* contains filtered or unexported fields */}
	VLast    Version = V2 // always last version
)

var (
	ErrCorrupted = errors.New("log corrupted")
)

var ErrInvalidOffset = errors.New("invalid offset")

var ErrNotFound = errors.New("not found")

var Invalid = Message{Offset: OffsetInvalid}

Functions

func MinOffset

func MinOffset(offsets map[int64]struct{}) int64

func Size

func Size(m Message, v Version) int64

Types

type Message

type Message struct {
	Offset int64
	Time   time.Time
	Key    []byte
	Value  []byte
}

func Gen

func Gen(count int) []Message

type Reader

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

func OpenReader

func OpenReader(path string, offset int64) (r *Reader, retErr error)

func OpenReaderMem

func OpenReaderMem(path string, offset int64) (r *Reader, retErr error)

func (*Reader) Close

func (r *Reader) Close() error

func (*Reader) Consume

func (r *Reader) Consume(position, maxPosition int64, maxCount int64) ([]Message, error)

func (*Reader) Get

func (r *Reader) Get(position int64) (msg Message, err error)

func (*Reader) InitialPosition

func (r *Reader) InitialPosition() int64

func (*Reader) Read

func (r *Reader) Read(position int64) (msg Message, nextPosition int64, err error)

func (*Reader) Version

func (r *Reader) Version() Version

type Version

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

func (Version) String

func (v Version) String() string

type Writer

type Writer struct {
	Path string
	// contains filtered or unexported fields
}

func OpenWriter

func OpenWriter(path string, offset int64, newVersion Version) (w *Writer, retErr error)

func (*Writer) Close

func (w *Writer) Close() error

func (*Writer) Size

func (w *Writer) Size() int64

func (*Writer) Sync

func (w *Writer) Sync() error

func (*Writer) SyncAndClose

func (w *Writer) SyncAndClose() error

func (*Writer) Version

func (w *Writer) Version() Version

func (*Writer) Write

func (w *Writer) Write(m Message) (int64, error)