Function New

engine/queue/queue.go:46–77

New instantiates a new queue. flushFunc is is used for periodic index flusing, and closeFunc will be used when closing. flushFunc should add items with values, and delete items without values. Nil items are possible. The goal is to batch index updates on a single thread.

(
	logger *zap.SugaredLogger,
	flushFunc func([]*Item) error,
	closeFunc func() error,
	opts Options,
)

//
// The goal is to batch index updates on a single thread.
func New(
	logger *zap.SugaredLogger,
	flushFunc func([]*Item) error,
	closeFunc func() error,
	opts Options,
) *Queue {
	if flushFunc == nil {
		flushFunc = func([]*Item) error { return nil }
	}
	if closeFunc == nil {
		closeFunc = func() error { return nil }
	}
	if opts.Rate == 0 {
		opts.Rate = 5 * time.Second
	}

	return &Queue{
		l: logger,

		closeFunc: closeFunc,
		flushFunc: flushFunc,

		pendingC:     make(chan *Item, opts.BatchSize),
		pending:      0,
		pendingItems: make([]*Item, opts.BatchSize),
		rate:         opts.Rate,
		batchSize:    opts.BatchSize,

		stopC:   make(chan bool, 1),
		stopped: true,
	}
}

// Queue indicates that a new item is pending insertion. A nil value indicates
// the item should be deleted.