source

func Consume ¶

func Consume(ctx context.Context, recv loki.LogsReceiver, f *loki.Fanout)

Consume continuously reads log entries from recv and forwards them to the fanout f. It runs until ctx is cancelled or an error occurs while sending to the fanout.

This function is typically used in component Run methods to handle the forwarding of log entries from a component's internal handler to downstream receivers. The fanout allows entries to be sent to multiple receivers concurrently.

func ConsumeBatch ¶

func ConsumeBatch(ctx context.Context, recv loki.LogsBatchReceiver, f *loki.Fanout)

ConsumeBatch continuously reads batches of log entries from recv and forwards them to the fanout f. It runs until ctx is cancelled or an error occurs while sending to the fanout.

This function is typically used in component Run methods to handle the forwarding of log entries from a component's internal handler to downstream receivers. The fanout allows entries to be sent to multiple receivers concurrently.

func Drain ¶

func Drain(recv loki.LogsReceiver, f func())

Drain consumes log entries from recv in a background goroutine while f executes. This prevents deadlocks that can occur when stopping components that may still be sending entries to the receiver channel. The draining goroutine will continue consuming entries until f returns, at which point the context is cancelled and the goroutine exits.

This is typically used during component shutdown to drain any remaining entries from a receiver channel while performing cleanup operations.

func Reconcile ¶

func Reconcile[Key comparable, Input any](
	logger log.Logger,
	s *Scheduler[Key],
	it iter.Seq[Input],
	keyFn KeyFn[Key, Input],
	sourceFactoryFn SourceFactoryFn[Key, Input],
)

Reconcile synchronizes the scheduler's set of running sources with a desired state. It iterates over inputs, creates sources for new items, and stops sources that are no longer needed.