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

• func ExecuteCh[T any](ctx context.Context, fns []ExecuteFunc[T]) chan ResultCh[T]
• func MapCh[T any, Result any](ctx context.Context, sl []T, f MapFunc[T, Result], opts ...Func) chan ResultCh[Result]
• type Errors
  • func Execute[T any](ctx context.Context, fns []ExecuteFunc[T]) ([]T, Errors)
  • func Map[T any, Result any](ctx context.Context, sl []T, f MapFunc[T, Result], opts ...Func) ([]Result, Errors)
  • func (e Errors) Error() string
• type ExecuteFunc
• type Func
  • func WithConcurrency(concurrency int) Func
• type MapFunc
• type Option
• type ResultCh

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 MapCh

func MapCh[T any, Result any](
	ctx context.Context,
	sl []T,
	f MapFunc[T, Result],
	opts ...Func,
) chan ResultCh[Result]

MapCh is a generic function in Go that applies a function concurrently to each element in a slice and sends the results via a channel. It uses a semaphore to control the concurrency level.

The function takes four parameters: - ctx: A context.Context which is used to control the cancellation of the computation. - concurrency: An int64 that represents the maximum number of concurrent goroutines that can run. If set to 0 or -1, it will use the runtime.NumCPU(). - sl: A slice of elements of type T. The function will be applied to each of these elements. - f: A MapFunc function that takes a context and an element of type T and returns a Result of type any and an error.

The function returns a channel which will receive a ResultCh struct containing three fields: - Index: The index of the input slice that the Result corresponds to. - Output: The Result of applying the function f to the corresponding element of the input slice. - Error: Any error that occurred while applying the function f.

The function uses a semaphore to limit the number of goroutines that can run concurrently. Each goroutine applies the function f to an element of the input slice and sends the result via the resultsCh channel. If the context is cancelled or the semaphore cannot be acquired, an error is sent via the resultsCh channel. If an error occurs while applying the function f, the error is also sent via the resultsCh channel.

NOTE: that the function does not close the resultsCh channel. It is the responsibility of the caller to ensure that all results are read from the channel to avoid a goroutine leak.

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,
	sl []T,
	f MapFunc[T, Result],
	opts ...Func,
) ([]Result, Errors)

Map calls the `Func` concurrently on each element of `sl`, and returns the results and any errors that occurred.

NOTE: For more info. see the `MapCh` function.

NOTE: Order is preserved.

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 WithConcurrency

func WithConcurrency(concurrency int) Func

WithConcurrency sets the OnFinished function.

type MapFunc

type MapFunc[T any, Result any] func(context.Context, 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 Option

type Option struct {
	// Concurrency is the number of concurrent goroutines that will be used.
	Concurrency int
}

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.