Documentation
¶
Index ¶
- Constants
- Variables
- func Backup(src, dst string) error
- func Check(dir string, opts Options) error
- func DeleteMulti(ctx context.Context, l Log, offsets map[int64]struct{}, ...) (map[int64]struct{}, int64, error)
- func Recover(dir string, opts Options) error
- type BlockingLog
- type Codec
- type DeleteMultiBackoff
- type JsonCodec
- type Log
- type Message
- type Options
- type Stats
- type TBlockingLog
- type TLog
- type TMessage
Constants ¶
const ( // OffsetOldest represents the smallest offset still available // Use it to consume all messages, starting at the first available OffsetOldest = message.OffsetOldest // OffsetNewest represents the offset that will be used for the next publish // Use it to consume only new messages OffsetNewest = message.OffsetNewest // OffsetInvalid is the offset returned when error is detected OffsetInvalid = message.OffsetInvalid )
Variables ¶
var ErrInvalidOffset = message.ErrInvalidOffset
ErrInvalidOffset error is returned when the offset attribute is invalid or out of bounds
var ErrNoIndex = errors.New("no index")
ErrNoIndex error is returned when we try to use key or timestamp, but the log doesn't include index on them
var ErrNotFound = message.ErrNotFound
ErrNotFound error is returned when the offset, key or timestamp is not found
var ErrReadonly = errors.New("log opened in readonly mode")
ErrReadonly error is returned when attempting to modify (e.g. publish or delete) from a log that is open as a readonly
var InvalidMessage = message.Invalid
InvalidMessage returned when an error has occurred
var StringCodec = stringCodec{}
StringCodec supports coding a string
var StringOptCodec = stringOptCodec{}
StringOptCodec supports coding an optional string, e.g. differentiates between "" and nil strings
var VarintCodec = varintCodec{}
VarintCodec supports coding integers as varint
Functions ¶
func DeleteMulti ¶ added in v0.3.0
func DeleteMulti(ctx context.Context, l Log, offsets map[int64]struct{}, backoff DeleteMultiBackoff) (map[int64]struct{}, int64, error)
DeleteMulti tries to delete all messages with offsets from the log and returns the amount of storage deleted
If error is encountered, it will return the deleted offsets and size, together with the error
DeleteMultiBackoff is called on each iteration to give others a chance to work with the log, while being deleted
Types ¶
type BlockingLog ¶ added in v0.6.0
type BlockingLog interface {
Log
// ConsumeBlocking is similar to [Consume], but if offset is equal to the next offset it will block until next message is published
ConsumeBlocking(ctx context.Context, offset int64, maxCount int64) (nextOffset int64, messages []Message, err error)
// ConsumeByKeyBlocking is similar to [ConsumeBlocking], but only returns messages matching the key
ConsumeByKeyBlocking(ctx context.Context, key []byte, offset int64, maxCount int64) (nextOffset int64, messages []Message, err error)
}
BlockingLog enhances Log adding blocking consume
func OpenBlocking ¶ added in v0.6.0
func OpenBlocking(dir string, opts Options) (BlockingLog, error)
OpenBlocking opens a Log and wraps it with support for blocking consume
func WrapBlocking ¶ added in v0.6.0
func WrapBlocking(l Log) (BlockingLog, error)
WrapBlocking wraps a Log with support for blocking consume
type Codec ¶ added in v0.5.0
type Codec[T any] interface { Encode(t T, empty bool) (b []byte, err error) Decode(b []byte) (t T, empty bool, err error) }
Codec is interface satisfied by all codecs
type DeleteMultiBackoff ¶ added in v0.3.0
DeleteMultiBackoff is called on each iteration of DeleteMulti to give applications opportunity to not overload the target log with deletes
func DeleteMultiWithWait ¶ added in v0.3.0
func DeleteMultiWithWait(d time.Duration) DeleteMultiBackoff
DeleteMultiWithWait returns a backoff func that sleeps/waits for a certain duration. If context is canceled while executing it returns the associated error
type JsonCodec ¶ added in v0.5.0
type JsonCodec[T any] struct{}
JsonCodec supports coding values as JSON
type Log ¶
type Log interface {
// Publish appends messages to the log.
// It returns the offset of the next message to be appended.
// The offset of the message is ignored, set to the actual offset.
// If the time of the message is 0, it is set to the current UTC time.
Publish(messages []Message) (nextOffset int64, err error)
// NextOffset returns the offset of the next message to be published.
NextOffset() (nextOffset int64, err error)
// Consume retrieves messages from the log, starting at the offset.
// It returns offset, which can be used to retrieve for the next consume.
// If offset == OffsetOldest, the first message will be the oldest
// message still available on the log. If the log is empty,
// it will return no error, nextOffset will be 0
// If offset == OffsetNewest, no actual messages will be returned,
// but nextOffset will be set to the offset that will be used
// for the next Publish call
// If offset is before the first available on the log, or is after
// NextOffset, it returns ErrInvalidOffset
// If the exact offset is already deleted, it will start consuming
// from the next available offset.
// Consume is allowed to return no messages, but with increasing nextOffset
// in case messages between offset and nextOffset have been deleted.
// NextOffset is always bigger than offset, unless we are caught up
// to the head of the log in which case they are equal.
Consume(offset int64, maxCount int64) (nextOffset int64, messages []Message, err error)
// ConsumeByKey is similar to Consume, but only returns messages matching the key
ConsumeByKey(key []byte, offset int64, maxCount int64) (nextOffset int64, messages []Message, err error)
// Get retrieves a single message, by its offset
// If offset == OffsetOldest, it returns the first message on the log
// If offset == OffsetNewest, it returns the last message on the log
// If offset is before the first available on the log, or is after
// NextOffset, it returns ErrInvalidOffset
// If log is empty, it returns ErrInvalidOffset
// If the exact offset has been deleted, it returns ErrNotFound
Get(offset int64) (message Message, err error)
// GetByKey retrieves the last message in the log for this key
// If no such message is found, it returns ErrNotFound
GetByKey(key []byte) (message Message, err error)
// OffsetByKey retrieves the last message offset in the log for this key
// If no such message is found, it returns ErrNotFound
OffsetByKey(key []byte) (offset int64, err error)
// GetByTime retrieves the first message after start time
// If start time is after all messages in the log, it returns ErrNotFound
GetByTime(start time.Time) (message Message, err error)
// OffsetByTime retrieves the first message offset and its time after start time
// If start time is after all messages in the log, it returns ErrNotFound
OffsetByTime(start time.Time) (offset int64, messageTime time.Time, err error)
// Delete tries to delete a set of messages by their offset
// from the log and returns the amount of storage deleted
// It does not guarantee that it will delete all messages,
// it returns the set of actually deleted offsets.
Delete(offsets map[int64]struct{}) (deletedOffsets map[int64]struct{}, deletedSize int64, err error)
// Size returns the amount of storage associated with a message
Size(m Message) int64
// Stat returns log stats like disk space, number of messages
Stat() (Stats, error)
// Backup takes a backup snapshot of this log to another location
Backup(dir string) error
// Sync forces persisting data to the disk. It returns the nextOffset
// at the time of the Sync, so clients can determine what portion
// of the log is now durable.
Sync() (nextOffset int64, err error)
// GC releases any unused resources associated with this log
GC(unusedFor time.Duration) error
// Close closes the log
Close() error
}
type Options ¶
type Options struct {
// When set will try to create all directories
CreateDirs bool
// Open the store in readonly mode
Readonly bool
// Index message keys, enabling GetByKey and OffsetByKey
KeyIndex bool
// Index message times, enabling GetByTime and OffsetByTime
TimeIndex bool
// Force filesystem sync after each Publish
AutoSync bool
// At what segment size it will rollover to a new segment. Defaults to 1MB.
Rollover int64
// Check the head segment for integrity, before opening it for reading/writing.
Check bool
}
type TBlockingLog ¶ added in v0.6.0
type TBlockingLog[K any, V any] interface { TLog[K, V] // ConsumeBlocking see [BlockingLog.ConsumeBlocking] ConsumeBlocking(ctx context.Context, offset int64, maxCount int64) (nextOffset int64, messages []TMessage[K, V], err error) // ConsumeByKeyBlocking see [BlockingLog.ConsumeByKeyBlocking] ConsumeByKeyBlocking(ctx context.Context, key K, empty bool, offset int64, maxCount int64) (nextOffset int64, messages []TMessage[K, V], err error) }
TBlockingLog enhances TLog adding blocking consume
func OpenTBlocking ¶ added in v0.6.0
func OpenTBlocking[K any, V any](dir string, opts Options, keyCodec Codec[K], valueCodec Codec[V]) (TBlockingLog[K, V], error)
OpenTBlocking opens a TLog and wraps it with support for blocking consume
func WrapTBlocking ¶ added in v0.6.0
func WrapTBlocking[K any, V any](l TLog[K, V]) (TBlockingLog[K, V], error)
WrapTBlocking wraps a TLog with support for blocking consume
type TLog ¶ added in v0.6.0
type TLog[K any, V any] interface { // Publish see [Log.Publish] Publish(messages []TMessage[K, V]) (nextOffset int64, err error) // NextOffset see [Log.NextOffset] NextOffset() (nextOffset int64, err error) // Consume see [Log.Consume] Consume(offset int64, maxCount int64) (nextOffset int64, messages []TMessage[K, V], err error) // ConsumeByKey see [Log.ConsumeByKey] ConsumeByKey(key K, empty bool, offset int64, maxCount int64) (nextOffset int64, messages []TMessage[K, V], err error) // Get see [Log.Get] Get(offset int64) (message TMessage[K, V], err error) // GetByKey see [Log.GetByKey] GetByKey(key K, empty bool) (message TMessage[K, V], err error) // OffsetByKey see [Log.OffsetByKey] OffsetByKey(key K, empty bool) (int64, error) // GetByTime see [Log.GetByTime] GetByTime(start time.Time) (message TMessage[K, V], err error) // OffsetByTime see [Log.OffsetByTime] OffsetByTime(start time.Time) (offset int64, messageTime time.Time, err error) // Delete see [Log.Delete] Delete(offsets map[int64]struct{}) (deletedOffsets map[int64]struct{}, deletedSize int64, err error) // Size see [Log.Size] Size(m Message) int64 // Stat see [Log.Stat] Stat() (Stats, error) // Backup see [Log.Backup] Backup(dir string) error // Sync see [Log.Sync] Sync() (nextOffset int64, err error) // GC see [Log.GC] GC(unusedFor time.Duration) error // Close see [Log.Close] Close() error // Raw returns the underlying log Raw() Log }
TLog is a typed Log which encodes/decodes keys and values to bytes.