storekit

package
v1.5.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 6, 2022 License: Apache-2.0 Imports: 9 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrInvalidData .
	ErrInvalidData = errors.New("invalid data")
	// ErrIteratorClosed .
	ErrIteratorClosed = errors.New("iterator closed")
	// ErrWriterClosed .
	ErrWriterClosed = errors.New("writer closed")
	// ErrOpNotSupported .
	ErrOpNotSupported = errors.New("operation not supported")
)
View Source 
var DefaultLatencyBuckets = []float64{1, 50, 100, 250, 500, 750, 1000, 5000, 10000}

Milliseconds

View Source 
var DefaultNopWriter = NopWriter{}

DefaultNopWriter .

View Source 
var DefaultStdoutWriter = StdoutWriter{
	Filter: func(val Data) bool { return true },
}

DefaultStdoutWriter .

View Source 
var ErrExitConsume = errors.New("exit consume")

ErrExitConsume .

Functions

func BatchConsume 

func BatchConsume(ctx context.Context, r BatchReader, w BatchWriter, opts *BatchConsumeOptions) error

BatchConsume .

Types

type BatchConsumeOptions 

type BatchConsumeOptions struct {
	BufferSize  int
	ReadTimeout time.Duration
	// ReadErrorHandler return err to exit, return nil to continue
	ReadErrorHandler func(err error) error
	// WriteErrorHandler return err to retry write, return nil to continue
	WriteErrorHandler func(list []Data, err error) error
	// ConfirmErrorHandler return err to exit, return nil to continue
	ConfirmErrorHandler func(err error) error
	Backoff             TimerBackoff
	Statistics          ConsumeStatistics
}

BatchConsumeOptions .

type BatchReader 

type BatchReader interface {
	ReadN(buf []Data, timeout time.Duration) (int, error)
	Confirm() error
	Close() error
}

BatchReader .

func NewMockBatchReader 

func NewMockBatchReader(ctx context.Context, interval time.Duration, batchSize int, creator func() interface{}) BatchReader

NewMockBatchReader .

type BatchWriter 

type BatchWriter interface {
	WriteN(vals ...Data) (int, error)
	Close() error
}

BatchWriter .

type BufferedWriter 

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

BufferedWriter .

func NewBufferedWriter 

func NewBufferedWriter(w BatchWriter, capacity int) *BufferedWriter

NewBufferedWriter .

func (*BufferedWriter) Close 

func (b *BufferedWriter) Close() error

Close .

func (*BufferedWriter) Data 

func (b *BufferedWriter) Data() []Data

Data .

func (*BufferedWriter) Flush 

func (b *BufferedWriter) Flush() error

Flush .

func (*BufferedWriter) Size 

func (b *BufferedWriter) Size() int

Size .

func (*BufferedWriter) Write 

func (b *BufferedWriter) Write(data Data) error

Write .

func (*BufferedWriter) WriteN 

func (b *BufferedWriter) WriteN(data ...Data) (int, error)

WriteN returns the number of buffers written to the data. if a Flush error occurs, the error will be returned

type Comparer 

type Comparer interface {
	Compare(a, b Data) int
}

Comparer .

type ConsumeStatistics 

type ConsumeStatistics interface {
	ReadError(err error)
	WriteError(data []Data, err error)
	ConfirmError(data []Data, err error)
	Success(data []Data)
	ObserveReadLatency(start time.Time)
	ObserveWriteLatency(start time.Time)
}

ConsumeStatistics . 

var NopConsumeStatistics ConsumeStatistics = &nopConsumeStatistics{}

type Counter added in v1.5.0 

type Counter interface {
	Total() (int64, error)
}

type Data 

type Data = interface{}

Data .

type EmptyIterator 

type EmptyIterator struct{}

EmptyIterator .

func (EmptyIterator) Close 

func (it EmptyIterator) Close() error

Close .

func (EmptyIterator) Error 

func (it EmptyIterator) Error() error

Error .

func (EmptyIterator) First 

func (it EmptyIterator) First() bool

First .

func (EmptyIterator) Last 

func (it EmptyIterator) Last() bool

Last .

func (EmptyIterator) Next 

func (it EmptyIterator) Next() bool

Next .

func (EmptyIterator) Prev 

func (it EmptyIterator) Prev() bool

Prev .

func (EmptyIterator) Total added in v1.5.0 

func (it EmptyIterator) Total() (int64, error)

Total .

func (EmptyIterator) Value 

func (it EmptyIterator) Value() interface{}

Value .

type Flusher 

type Flusher interface {
	Flush() error
}

Flusher

type Int64Comparer 

type Int64Comparer struct{}

Int64Comparer .

func (Int64Comparer) Compare 

func (c Int64Comparer) Compare(a, b Data) int

Compare .

type Iterator 

type Iterator interface {
	First() bool
	Last() bool
	Next() bool
	Prev() bool
	Value() Data
	Error() error
	Close() error
}

Iterator .

func MergedHeadOverlappedIterator 

func MergedHeadOverlappedIterator(cmp Comparer, its ...Iterator) Iterator

MergedHeadOverlappedIterator .

func NewListIterator 

func NewListIterator(list ...Data) Iterator

NewListIterator .

func OrderedIterator added in v1.5.0 

func OrderedIterator(its ...Iterator) Iterator

OrderedIterator .

type ListIterator 

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

ListIterator .

func (*ListIterator) Close 

func (it *ListIterator) Close() error

Close .

func (*ListIterator) Error 

func (it *ListIterator) Error() error

Error .

func (*ListIterator) First 

func (it *ListIterator) First() bool

First .

func (*ListIterator) Last 

func (it *ListIterator) Last() bool

Last .

func (*ListIterator) Next 

func (it *ListIterator) Next() bool

Next .

func (*ListIterator) Prev 

func (it *ListIterator) Prev() bool

Prev .

func (*ListIterator) Total added in v1.5.0 

func (it *ListIterator) Total() (int64, error)

Total .

func (*ListIterator) Value 

func (it *ListIterator) Value() Data

Value .

type Matcher 

type Matcher interface {
	Match(val Data) bool
}

Matcher .

type MultiplierBackoff 

type MultiplierBackoff struct {
	Base     time.Duration
	Max      time.Duration
	Duration time.Duration
	Factor   float64
}

MultiplierBackoff .

func NewMultiplierBackoff 

func NewMultiplierBackoff() *MultiplierBackoff

NewMultiplierBackoff

func (*MultiplierBackoff) Reset 

func (b *MultiplierBackoff) Reset()

func (*MultiplierBackoff) Wait 

func (b *MultiplierBackoff) Wait() <-chan time.Time

type NopWriter 

type NopWriter struct{}

NopWriter .

func (NopWriter) Close 

func (w NopWriter) Close() error

func (NopWriter) Write 

func (w NopWriter) Write(val Data) error

func (NopWriter) WriteN 

func (w NopWriter) WriteN(vals ...Data) (int, error)

type StdoutWriter 

type StdoutWriter struct {
	Filter func(val Data) bool
}

StdoutWriter .

func (StdoutWriter) Close 

func (w StdoutWriter) Close() error

func (StdoutWriter) Write 

func (w StdoutWriter) Write(val Data) error

func (StdoutWriter) WriteN 

func (w StdoutWriter) WriteN(vals ...Data) (int, error)

type TimerBackoff 

type TimerBackoff interface {
	Reset()
	Wait() <-chan time.Time
}

TimerBackoff .

type Writer 

type Writer interface {
	Write(val Data) error
	Close() error
}

Writer .

func WrapBatchWriter 

func WrapBatchWriter(
	bw BatchWriter,
	bufferSize uint64,
	timeout time.Duration,
	errorh func(error) error,
) Writer

WrapBatchWriter .

Source Files

View all Source files

Directories

Path Synopsis
index
index/cleaner
index/cleaner/retention-default
index/creator
index/initializer
index/loader
index/retention-strategy
index/rollover
kafka
topic/initializer

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.