parsequeue

Documentation

Index

• type AckFunc
• type ParseFunc
• type ParseQueue
  • func New[TData any](lgr log.Logger, parallelism int, sink abstract.AsyncSink, ...) *ParseQueue[TData]
  • func (p *ParseQueue[TData]) Add(message TData) error
  • func (p *ParseQueue[TData]) Close()
• type WaitableParseQueue
  • func NewWaitable[TData any](lgr log.Logger, parallelism int, sink abstract.AsyncSink, ...) *WaitableParseQueue[TData]
  • func (p *WaitableParseQueue[TData]) Add(message TData) error
  • func (p *WaitableParseQueue[TData]) Close()
  • func (p *WaitableParseQueue[TData]) Wait()

Types

type AckFunc

type AckFunc[TData any] func(data TData, pushSt time.Time, err error)

type ParseFunc

type ParseFunc[TData any] func(TData) []abstract.ChangeItem

type ParseQueue

type ParseQueue[TData any] struct {
	// contains filtered or unexported fields
}

func New

func New[TData any](
	lgr log.Logger,
	parallelism int,
	sink abstract.AsyncSink,
	parseF ParseFunc[TData],
	ackF AckFunc[TData],
) *ParseQueue[TData]

func (*ParseQueue[TData]) Add

func (p *ParseQueue[TData]) Add(message TData) error

Add will schedule new message parse

Do not call concurrently with Close()!

func (*ParseQueue[TData]) Close

func (p *ParseQueue[TData]) Close()

Close shutdown all goroutines

Do not call concurrently with Add()

type WaitableParseQueue

type WaitableParseQueue[TData any] struct {
	// contains filtered or unexported fields
}

func NewWaitable

func NewWaitable[TData any](
	lgr log.Logger,
	parallelism int,
	sink abstract.AsyncSink,
	parseF ParseFunc[TData],
	ackF AckFunc[TData],
) *WaitableParseQueue[TData]

func (*WaitableParseQueue[TData]) Add

func (p *WaitableParseQueue[TData]) Add(message TData) error

func (*WaitableParseQueue[TData]) Close

func (p *WaitableParseQueue[TData]) Close()

func (*WaitableParseQueue[TData]) Wait

func (p *WaitableParseQueue[TData]) Wait()

Wait waits when all messages, added via .Add() will be acked

Should be called mutually exclusive with Add()/Close()