util

Documentation

Index

• func AllClosed(channels ...<-chan struct{}) <-chan struct{}
• func AllDone(components ...module.ReadyDoneAware) <-chan struct{}
• func AllReady(components ...module.ReadyDoneAware) <-chan struct{}
• func CheckClosed(done <-chan struct{}) bool
• func DetypeSlice[T any](typedSlice []T) []any
• func IsEmptyOrNotExists(path string) (bool, error)
• func MergeChannels(channels interface{}) interface{}
• func MergeReadyDone(components ...module.ReadyDoneAware) module.ReadyDoneAware
• func SampleN(n int, max, percentage float64) uint
• func WaitClosed(ctx context.Context, ch <-chan struct{}) error
• func WaitError(errChan <-chan error, done <-chan struct{}) error
• type LogProgressConfig
  • func DefaultLogProgressConfig[T int | uint | int32 | uint32 | uint64 | int64](message string, total T) LogProgressConfig[T]
  • func NewLogProgressConfig[T int | uint | int32 | uint32 | uint64 | int64](message string, total T, noDataLogDuration time.Duration, ticks uint64) LogProgressConfig[T]
• type LogProgressFunc
  • func LogProgress[T int | uint | int32 | uint32 | uint64 | int64](log zerolog.Logger, config LogProgressConfig[T]) LogProgressFunc[T]

Functions

func AllClosed

func AllClosed(channels ...<-chan struct{}) <-chan struct{}

AllClosed returns a channel that is closed when all input channels are closed.

func AllDone

func AllDone(components ...module.ReadyDoneAware) <-chan struct{}

AllDone calls Done on all input components and returns a channel that is closed when all input components are done.

func AllReady

func AllReady(components ...module.ReadyDoneAware) <-chan struct{}

AllReady calls Ready on all input components and returns a channel that is closed when all input components are ready.

func CheckClosed

func CheckClosed(done <-chan struct{}) bool

CheckClosed checks if the provided channel has a signal or was closed. Returns true if the channel was signaled/closed, otherwise, returns false.

This is intended to reduce boilerplate code when multiple channel checks are required because missed signals could cause safety issues.

func DetypeSlice

func DetypeSlice[T any](typedSlice []T) []any

DetypeSlice converts a typed slice containing any kind of elements into an untyped []any type, in effect removing the element type information from the slice. It is useful for passing data into structpb.NewValue, which accepts []any but not []T for any specific type T.

func IsEmptyOrNotExists

func IsEmptyOrNotExists(path string) (bool, error)

IsEmptyOrNotExists returns true if the directory does not exist or is empty. It returns an error if there's an issue accessing the directory.

func MergeChannels

func MergeChannels(channels interface{}) interface{}

MergeChannels merges a list of channels into a single channel

func MergeReadyDone

func MergeReadyDone(components ...module.ReadyDoneAware) module.ReadyDoneAware

MergeReadyDone merges []module.ReadyDoneAware into one module.ReadyDoneAware.

func SampleN

func SampleN(n int, max, percentage float64) uint

SampleN computes a percentage of the given number 'n', and returns the result as an unsigned integer. If the calculated sample is greater than the provided 'max' value, it returns the ceil of 'max'. If 'n' is less than or equal to 0, it returns 0.

Parameters: - n: The input number, used as the base to compute the percentage. - max: The maximum value that the computed sample should not exceed. - percentage: The percentage (in range 0.0 to 1.0) to be applied to 'n'.

Returns: - The computed sample as an unsigned integer, with consideration to the given constraints.

func WaitClosed

func WaitClosed(ctx context.Context, ch <-chan struct{}) error

WaitClosed waits for either a signal/close on the channel or for the context to be cancelled Returns nil if the channel was signalled/closed before returning, otherwise, it returns the context error.

This handles the corner case where the context is cancelled at the same time that the channel is closed, and the Done case was selected. This is intended for situations where ignoring a signal can cause safety issues.

func WaitError

func WaitError(errChan <-chan error, done <-chan struct{}) error

WaitError waits for either an error on the error channel or the done channel to close Returns an error if one is received on the error channel, otherwise it returns nil

This handles a race condition where the done channel could have been closed as a result of an irrecoverable error being thrown, so that when the scheduler yields control back to this goroutine, both channels are available to read from. If the done case happens to be chosen at random to proceed instead of the error case, then we would return without error which could result in unsafe continuation.

Types

type LogProgressConfig

type LogProgressConfig[T int | uint | int32 | uint32 | uint64 | int64] struct {
	// contains filtered or unexported fields
}

func DefaultLogProgressConfig

func DefaultLogProgressConfig[T int | uint | int32 | uint32 | uint64 | int64](
	message string,
	total T,
) LogProgressConfig[T]

DefaultLogProgressConfig returns a LogProgressConfig with default values. The default values will log every 10% and will log an additional line if new data is received after no data has been received for 1 minute.

func NewLogProgressConfig

func NewLogProgressConfig[T int | uint | int32 | uint32 | uint64 | int64](
	message string,
	total T,
	noDataLogDuration time.Duration,
	ticks uint64,
) LogProgressConfig[T]

NewLogProgressConfig creates and returns a new LogProgressConfig with the specified message, total, duration, and ticks. The duration is rounded to the nearest millisecond. The number of ticks is the number of increments to log at. Logging at 0% is always done. If you want to log at 10% increments, set ticks to 10.

type LogProgressFunc

type LogProgressFunc[T int | uint | int32 | uint32 | uint64 | int64] func(addProgress T)

LogProgressFunc is a function that can be called to add to the progress. The function can be called concurrently. addProgress is the amount to add to the progress. It is any integer number type, but all negative values are ignored.

func LogProgress

func LogProgress[T int | uint | int32 | uint32 | uint64 | int64](
	log zerolog.Logger,
	config LogProgressConfig[T],
) LogProgressFunc[T]

LogProgress takes a LogProgressConfig and return function such that when called adds the given number to the progress and logs the progress in defined increments or there is a time gap between progress updates. The returned function can be called concurrently. An eta is also logged, but it assumes that the progress is linear.