quasar

Apr 4, 2020

Changes in this version

type DB

+ func (db *DB) Close() error

Jul 15, 2019

Changes in this version

type DBConfig

+ SyncWrites bool

type Ledger

+ func (l *Ledger) Tail() uint64

Jul 12, 2019

Changes in this version

type DBConfig

+ GCRatio float64

Jul 8, 2019

Changes in this version

+ func EncodeSequenceTo(w io.Writer, s uint64, compact bool) error

type Buffer

+ func (b *Buffer) Index(index int) (Entry, bool)

type QueueConfig

+ CleanErrors func(error)

+ CleanInterval time.Duration

+ CleanRetention int

+ CleanThreshold int

+ LedgerCache int

+ LedgerLimit int

+ TableCache bool

type TableConfig

+ Cache bool

Jul 7, 2019

Changes in this version

type DBConfig

+ SyncErrors func(error)

+ SyncInterval time.Duration

Jul 6, 2019

Changes in this version

type Table

+ func (m *Table) All() (map[string][]uint64, error)

Jul 6, 2019

Changes in this version

+ const EncodedSequenceLength

+ var ErrConsumerDeadlock = errors.New("consumer deadlock")

type ConsumerConfig

+ Deadline time.Duration

Jul 5, 2019

Changes in this version

type DBConfig

+ Logger func(level Level, format string, args ...interface{})

+ type Level int

+ const Debug

+ const Error

+ const Info

+ const Warning

Jul 5, 2019

Changes in this version

+ var ErrConsumerTimeout = errors.New("consumer timeout")

type ConsumerConfig

+ Timeout time.Duration

Jul 4, 2019

Changes in this version

+ var ErrConsumerClosed = errors.New("consumer closed")

+ var ErrProducerClosed = errors.New("producer closed")

+ type Queue struct

+ func CreateQueue(db *DB, config QueueConfig) (*Queue, error)

+ func (q *Queue) Close()

+ func (q *Queue) Consumer(config ConsumerConfig) *Consumer

+ func (q *Queue) Ledger() *Ledger

+ func (q *Queue) Producer(config ProducerConfig) *Producer

+ func (q *Queue) Table() *Table

+ type QueueConfig struct

+ Cache int

+ Errors func(error)

+ Interval time.Duration

+ Limit int

+ Prefix string

+ Retention int

+ Threshold int

Jul 1, 2019

Changes in this version

type ProducerConfig

+ Filter bool

Jul 1, 2019

Changes in this version

type Entry

+ Object interface{}

Jul 1, 2019

Changes in this version

+ func CompileSequences(table map[uint64]bool) []uint64

+ type Buffer struct

+ func NewBuffer(size int) *Buffer

+ func (b *Buffer) Length() int

+ func (b *Buffer) Push(entries ...Entry)

+ func (b *Buffer) Reset()

+ func (b *Buffer) Scan(fn func(Entry) bool)

+ func (b *Buffer) Trim(fn func(Entry) bool)

type Consumer

+ func (c *Consumer) Mark(sequence uint64, cumulative bool, ack func(error))

type ConsumerConfig

+ Start uint64

+ Window int

type ProducerConfig

+ Delay time.Duration

+ Retry int

Jun 26, 2019

Changes in this version

+ var ErrLimitReached = errors.New("limit reached")

type CleanerConfig

+ Interval time.Duration

+ Retention int

+ Threshold int

type DBConfig

+ GCErrors chan<- error

type LedgerConfig

+ Limit int

type ProducerConfig

+ BatchSize int

+ BatchTimeout time.Duration

+ RetryInterval time.Duration

+ RetryTimeout time.Duration

Jun 25, 2019

Changes in this version

type DBConfig

+ LogSink io.Writer

Jun 25, 2019

Changes in this version

+ var ErrInvalidSequence = errors.New("invalid sequence")

type Cache

+ func (q *Cache) Trim(fn func(Entry) bool)

May 10, 2019

Changes in this version

type Matrix

+ func (m *Matrix) Clear() error

type Table

+ func (t *Table) Clear() error

May 9, 2019

Changes in this version

+ type Cache struct

+ func NewCache(size int) *Cache

+ func (q *Cache) Add(entries ...Entry)

+ func (q *Cache) Length() int

+ func (q *Cache) Reset()

+ func (q *Cache) Scan(fn func(int, Entry) bool)

type LedgerConfig

+ Cache int

May 9, 2019

Changes in this version

+ type CleanerConfig struct

+ Delay time.Duration

+ Errors chan<- error

+ Matrices []*Matrix

+ MaxRetention int

+ MinRetention int

+ Tables []*Table

+ type ConsumerConfig struct

+ Batch int

+ Entries chan<- Entry

+ Errors chan<- error

+ Name string

+ Skip int

+ type DBConfig struct

+ Errors chan<- error

+ GCInterval time.Duration

+ type LedgerConfig struct

+ Prefix string

+ type MatrixConfig struct

+ Prefix string

+ type ProducerConfig struct

+ Batch int

+ Timeout time.Duration

+ type ReaderConfig struct

+ Batch int

+ Entries chan<- Entry

+ Errors chan<- error

+ Start uint64

+ type TableConfig struct

+ Prefix string

+ type WorkerConfig struct

+ Batch int

+ Entries chan<- Entry

+ Errors chan<- error

+ Name string

+ Skip int

+ Window int

May 7, 2019

Changes in this version

+ func DecodeSequences(value []byte) ([]uint64, error)

+ func EncodeSequences(list []uint64) []byte

type CleanerOptions

+ Matrices []*Matrix

+ type Matrix struct

+ func CreateMatrix(db *DB, opts MatrixOptions) (*Matrix, error)

+ func (t *Matrix) Count() (int, error)

+ func (t *Matrix) Delete(name string) error

+ func (t *Matrix) Get(name string) ([]uint64, error)

+ func (t *Matrix) Range() (uint64, uint64, error)

+ func (t *Matrix) Set(name string, sequences []uint64) error

+ type MatrixOptions struct

+ Prefix string

+ type Worker struct

+ func NewWorker(ledger *Ledger, matrix *Matrix, opts WorkerOptions) *Worker

+ func (w *Worker) Ack(sequence uint64)

+ func (w *Worker) Close()

+ type WorkerOptions struct

+ Batch int

+ Entries chan<- Entry

+ Errors chan<- error

+ Name string

+ Window int

May 6, 2019

Changes in this version

+ const SequenceLength

+ var ErrNotMonotonic = errors.New("not monotonic")

+ func DecodeSequence(key []byte) (uint64, error)

+ func EncodeSequence(s uint64) []byte

+ func GenerateSequence(n uint32) uint64

+ func JoinSequence(ts time.Time, n uint32) uint64

+ func SplitSequence(s uint64) (time.Time, uint32)

+ type Cleaner struct

+ func NewCleaner(ledger *Ledger, opts CleanerOptions) *Cleaner

+ func (c *Cleaner) Close()

+ type CleanerOptions struct

+ Delay time.Duration

+ Errors chan<- error

+ MaxRetention int

+ MinRetention int

+ Tables []*Table

+ type Consumer struct

+ func NewConsumer(ledger *Ledger, table *Table, opts ConsumerOptions) *Consumer

+ func (c *Consumer) Ack(position uint64) error

+ func (c *Consumer) Close()

+ type ConsumerOptions struct

+ Batch int

+ Entries chan<- Entry

+ Errors chan<- error

+ Name string

+ type DB = badger.DB

+ func OpenDB(dir string, opts DBOptions) (*DB, error)

+ type DBOptions struct

+ Errors chan<- error

+ GCInterval time.Duration

+ type Entry struct

+ Payload []byte

+ Sequence uint64

+ type Ledger struct

+ func CreateLedger(db *DB, opts LedgerOptions) (*Ledger, error)

+ func (l *Ledger) Clear() error

+ func (l *Ledger) Delete(sequence uint64) error

+ func (l *Ledger) Head() uint64

+ func (l *Ledger) Index(index int) (uint64, error)

+ func (l *Ledger) Length() int

+ func (l *Ledger) Read(sequence uint64, amount int) ([]Entry, error)

+ func (l *Ledger) Reset() error

+ func (l *Ledger) Subscribe(receiver chan<- uint64)

+ func (l *Ledger) Unsubscribe(receiver chan<- uint64)

+ func (l *Ledger) Write(entries ...Entry) error

+ type LedgerOptions struct

+ Prefix string

+ type Producer struct

+ func NewProducer(ledger *Ledger, opts ProducerOptions) *Producer

+ func (p *Producer) Close()

+ func (p *Producer) Write(entry Entry, ack func(error)) bool

+ type ProducerOptions struct

+ Batch int

+ Timeout time.Duration

+ type Reader struct

+ func NewReader(ledger *Ledger, opts ReaderOptions) *Reader

+ func (r *Reader) Close()

+ type ReaderOptions struct

+ Batch int

+ Entries chan<- Entry

+ Errors chan<- error

+ Start uint64

+ type Table struct

+ func CreateTable(db *DB, opts TableOptions) (*Table, error)

+ func (t *Table) Count() (int, error)

+ func (t *Table) Delete(name string) error

+ func (t *Table) Get(name string) (uint64, error)

+ func (t *Table) Range() (uint64, uint64, error)

+ func (t *Table) Set(name string, position uint64) error

+ type TableOptions struct

+ Prefix string