concurrentloop

README

concurrentloop

concurrentloop provides a function to call a function concurrently powered by generics, channels, and goroutines.

Install

$ go get github.com/thalesfsp/concurrentloop

Specific version

Example: $ go get github.com/thalesfsp/concurrentloop@v1.2.0

Usage

See example_test.go, and concurrentloop_test.go file.

Documentation

Run $ make doc or check out online.

Development

Check out CONTRIBUTION.

Release

1. Update CHANGELOG accordingly.
2. Once changes from MR are merged.
3. Tag and release.

Roadmap

Check out CHANGELOG.

Documentation

Overview

Package concurrentloop provides a function to call a function concurrently powered by generics, channels, and goroutines.

Index

• Constants
• func ExecuteCh[T any](ctx context.Context, fns []ExecuteFunc[T]) chan ResultCh[T]
• func Flatten2D[T any](data [][]T) []T
• func RemoveZeroValues[T any](removeZeroValues bool, results []T) []T
• func SplitSlice[T any](items []T, batchSize int) [][]T
• type Errors
  • func Execute[T any](ctx context.Context, fns []ExecuteFunc[T]) ([]T, Errors)
  • func Map[T any, Result any](ctx context.Context, items []T, f MapFunc[T, Result], opts ...Func) ([]Result, Errors)
  • func MapCh[T any, Result any](ctx context.Context, itemsMap map[string]T, f MapMFuncCh[T, Result], ...) Errors
  • func MapDone[T any, Result any](ctx context.Context, items []T, f MapFuncCh[T, Result], opts ...Func) ([]Result, Errors)
  • func MapM[T any, Result any](ctx context.Context, itemsMap map[string]T, f MapMFunc[T, Result], ...) ([]Result, Errors)
  • func (e Errors) Error() string
• type ExecuteFunc
• type Func
  • func WithBatchSize(concurrency int) Func
  • func WithLimit(limit int) Func
  • func WithRandomDelayTime(minDelay, maxDelay int, d time.Duration) Func
  • func WithRemoveZeroValues(remove bool) Func
  • func WithWriter(w io.Writer) Func
• type MapFunc
• type MapFuncCh
• type MapMFunc
• type MapMFuncCh
• type Option
• type ResultCh

Constants

const Name = "concurrentloop"

Name of the package.

Functions

func ExecuteCh

func ExecuteCh[T any](ctx context.Context, fns []ExecuteFunc[T]) chan ResultCh[T]

ExecuteCh calls the `fns` concurrently.

NOTE: It's the caller's responsibility to close the channel.

func Flatten2D

func Flatten2D[T any](data [][]T) []T

Flatten2D takes a 2D slice and returns a 1D slice containing all the elements.

func RemoveZeroValues

func RemoveZeroValues[T any](removeZeroValues bool, results []T) []T

RemoveZeroValues removes zero values from the results.

func SplitSlice

func SplitSlice[T any](items []T, batchSize int) [][]T

SplitSlice splits a slice into batches of a given size.

Types

type Errors

type Errors []error

Errors is a slice of errors.

func Execute

func Execute[T any](ctx context.Context, fns []ExecuteFunc[T]) ([]T, Errors)

Execute calls the `fns` concurrently, and returns the results and any errors that occurred. The function blocks until all executions have completed.

NOTE: Order is may preserved.

func Map

func Map[T any, Result any](
	ctx context.Context,
	items []T,
	f MapFunc[T, Result],
	opts ...Func,
) ([]Result, Errors)

Map processes each element in a slice concurrently using the provided function. Returns a slice of results and any errors that occurred during processing.

func MapCh

func MapCh[T any, Result any](
	ctx context.Context,
	itemsMap map[string]T,
	f MapMFuncCh[T, Result],
	perCycleCh chan<- Result,
	endCh chan<- Result,
	opts ...Func,
) Errors

MapCh processes each key-value pair in a map concurrently with real-time result streaming. It follows the same pattern as Map but sends results to channels instead of returning them. Per-cycle results are sent to perCycleCh, final results (after RemoveZeroValues) are sent to endCh. Returns only errors that occurred during processing.

func MapDone

func MapDone[T any, Result any](
	ctx context.Context,
	items []T,
	f MapFuncCh[T, Result],
	opts ...Func,
) ([]Result, Errors)

MapDone processes each element in a slice concurrently with early termination support. The processing function receives a done channel that can be used to signal early termination. Returns a slice of results and any errors that occurred during processing.

func MapM

func MapM[T any, Result any](
	ctx context.Context,
	itemsMap map[string]T,
	f MapMFunc[T, Result],
	opts ...Func,
) ([]Result, Errors)

MapM processes each key-value pair in a map concurrently using the provided function. Returns a slice of results and any errors that occurred during processing.

func (Errors) Error

func (e Errors) Error() string

Error returns a string representation of the combined errors in the `Errors` slice, separated by commas. This method satisfies the `error` interface.

type ExecuteFunc

type ExecuteFunc[T any] func(context.Context) (T, error)

ExecuteFunc is the type of the function that will be executed concurrently for each element in a slice of type `T`. The function takes a `context.Context` and a value of type `T`, and returns a value of type `Result` and an error value.

type Func

type Func func(o Option) Option

Func allows to specify message's options.

func WithBatchSize

func WithBatchSize(concurrency int) Func

WithBatchSize sets the size of the batch.

func WithLimit

func WithLimit(limit int) Func

WithLimit sets the max amount of results to collect before stopping the loop.

func WithRandomDelayTime

func WithRandomDelayTime(minDelay, maxDelay int, d time.Duration) Func

WithRandomDelayTime sets the random delay time between each iteration.

func WithRemoveZeroValues

func WithRemoveZeroValues(remove bool) Func

WithRemoveZeroValues if set to true removes zero values from the results.

func WithWriter

func WithWriter(w io.Writer) Func

WithWriter sets a writer to write results to.

type MapFunc

type MapFunc[T any, Result any] func(ctx context.Context, item T) (Result, error)

MapFunc is the type of the function that will be executed concurrently for each element in a slice of type `T`. The function takes a `context.Context` and a value of type `T`, and returns a value of type `Result` and an error value.

type MapFuncCh

type MapFuncCh[T any, Result any] func(ctx context.Context, item T, done chan<- struct{}) (Result, error)

MapFuncCh is the type of the function that will be executed concurrently for each element in a slice with a done channel for early termination.

type MapMFunc

type MapMFunc[T any, Result any] func(ctx context.Context, key string, item T) (Result, error)

MapMFunc is the type of the function that will be executed concurrently for each element in the map.

type MapMFuncCh

type MapMFuncCh[T any, Result any] func(
	ctx context.Context,
	key string,
	item T,
	perCycleCh chan<- Result,
	endCh chan<- Result,
) (Result, error)

MapMFuncCh is the type of the function that will be executed concurrently for each element in a map with per-cycle and end channels for real-time result streaming.

type Option

type Option struct {
	// BatchSize is the size of the batch.
	BatchSize int

	// The max amount of results to collect before
	Limit int

	// RandomDelayTimeDuration is the unit of the duration (Second, Millisecond, etc.)
	RandomDelayTimeDuration time.Duration

	// RandomDelayTimeMax is the upper limit.
	RandomDelayTimeMax int

	// RandomDelayTimeMin is the lower limit.
	RandomDelayTimeMin int

	// RemoveZeroValues indicates whether to remove zero values from the results.
	RemoveZeroValues bool

	// Writer is an optional writer to write results to.
	Writer io.Writer
}

Option for the concurrent loop.

type ResultCh

type ResultCh[T any] struct {
	Error  error
	Index  int
	Output T
}

ResultCh receives the result from the channel.