Method Add

esutil/bulk_indexer.go:369–418

Add adds an item to the indexer. Adding an item after a call to Close() will panic.

(ctx context.Context, item BulkIndexerItem)

//
// Adding an item after a call to Close() will panic.
func (bi *bulkIndexer) Add(ctx context.Context, item BulkIndexerItem) error {
	atomic.AddUint64(&bi.stats.numAdded, 1)

	if item.ctx == nil {
		item.ctx = ctx
	}

	// Serialize metadata to JSON
	item.marshallMeta()
	// Compute length for body & metadata
	if err := item.computeLength(); err != nil {
		return err
	}

	n := bi.addCounter.Add(1) - 1
	numWorkers := uint64(len(bi.workers))
	primary := bi.workers[n%numWorkers]

	// Avoid the O(N) scan when the hashed worker can take the item immediately.
	select {
	case primary.ch <- item:
		return nil
	default:
	}

	// Spread load: prefer making progress on any idle worker over blocking
	// on one busy worker. Start after primary to avoid retrying it and to
	// distribute scan pressure evenly under sustained back-pressure.
	for i := uint64(1); i < numWorkers; i++ {
		w := bi.workers[(n+i)%numWorkers]
		select {
		case w.ch <- item:
			return nil
		default:
		}
	}

	// Back-pressure: all queues full; block until primary drains or the
	// context is cancelled.
	select {
	case <-ctx.Done():
		if bi.config.OnError != nil {
			bi.config.OnError(ctx, ctx.Err())
		}
		return ctx.Err()
	case primary.ch <- item:
	}

	return nil
}

// Close stops the periodic flush, closes the indexer queue channel,
// which triggers the workers to flush and stop.