cff

func EmitterStack ¶

func EmitterStack(emitters ...Emitter) Emitter

EmitterStack combines multiple emitters together into one.

Events are sent to the emitters in an unspecified order. Emitters should not assume the ordering of events.

func NopEmitter ¶

func NopEmitter() Emitter

NopEmitter is a cff emitter that does not do anything.

func NopFlowEmitter ¶

func NopFlowEmitter() FlowEmitter

NopFlowEmitter is a Flow emitter that does not do anything.

func MapEnd ¶

func MapEnd(fn interface{}) MapOption

MapEnd specifies a function for execution when a Map operation finishes. This function will run after all items in the map have been processed.

As with parallel tasks, the function passed to MapEnd may have:

• an optional context.Context parameter
• an optional error return value

Therefore, these are all valid:

cff.MapEnd(func() {...})
cff.MapEnd(func() error {...})
cff.MapEnd(func(ctx context.Context) {...})
cff.MapEnd(func(ctx context.Context) error {...})

MapEnd cannot be used with ContinueOnError.

This is a code generation directive.

func Concurrency ¶

func Concurrency(n int) Option

Concurrency specifies the maximum number of goroutines cff should use to execute tasks of this Flow or Parallel.

The default value for this is,

max(GOMAXPROCS, 4)

That is, by default cff will use runtime.GOMAXPROCS goroutines, with a minimum of 4.

This is a code generation directive.

func ContinueOnError ¶

func ContinueOnError(bool) Option

ContinueOnError configures a Parallel to keep running all other tasks despite errors returned by tasks over the course of its execution. By default, Parallel will stop execution at the first error it encounters.

err = cff.Parallel(ctx,
	cff.Task(task1),
	cff.Task(task2),
	// ...
	cff.ContinueOnError(true),
)

If one or more tasks return errors with ContinueOnError(true), Parallel will still run all the other tasks, and accumulate and combine the errors together into a single error object. You can access the full list of errors with go.uber.org/multierr.Errors.

ContinueOnError(true) is incompatible with Flow, SliceEnd and MapEnd.

This is a code generation directive.

func InstrumentFlow ¶

func InstrumentFlow(name string) Option

InstrumentFlow specifies that this Flow should be instrumented for observability. The provided name will be passed to the Emitter you passed into WithEmitter.

This is a code generation directive.

func InstrumentParallel ¶

func InstrumentParallel(name string) Option

InstrumentParallel specifies that this Parallel should be instrumented for observability. The provided name will be passed to the Emitter you passed into WithEmitter.

This is a code generation directive.

func Map ¶

func Map(fn interface{}, m interface{}, opts ...MapOption) Option

Map runs fn in parallel on elements of the provided map with a bounded number of goroutines.

cff.Parallel(ctx,
	cff.Map(
		func(k string, v *User) { /* ... */ },
		map[string]*User{ /* ... */ },
	),
)

For a slice map[K]V, fn has the following signature:

func([ctx context.Context,] k K, v V) ([error])

That is, it has the following parameters in-order:

• an optional context.Context
• a key in the map
• the value of that key in the map

And if the operation is fallible, it may have an error return value. A non-nil error returned by the function halts the entire Parallel operation. Use ContinueOnError to change this.

Map may only be used with Parallel.

This is a code generation directive.

func Params ¶

func Params(args ...interface{}) Option

Params specifies inputs for a Flow that do not have any dependencies. These values are made available to the Flow as-is.

For example:

var req *GetUserRequest = // ...
cff.Flow(
	cff.Params(req),
	// ...
)

This is a code generation directive.

func Results ¶

func Results(results ...interface{}) Option

Results specifies one or more outputs for a Flow. Arguments to Results must be pointers to variables that will hold the result values.

For example:

var result *GetUserResponse
err := cff.Flow(ctx,
	cff.Results(&result),
	// ...
)

This is a code generation directive.

func Slice ¶

func Slice(fn interface{}, slice interface{}, opts ...SliceOption) Option

Slice runs fn in parallel on elements of the provided slice with a bounded number of goroutines.

cff.Parallel(ctx,
	cff.Slice(
		func(el someType) { ... },
		[]someType{...},
	),
)

For a slice []T, fn has the following signature:

func([ctx context.Context,] [idx int,] value T) ([error])

That is, it has the following parameters in-order:

• an optional context.Context
• an optional integer holding the index of the element in the slice
• a value in the slice

And if the operation is fallible, it may have an error return value. A non-nil error returned by the function halts the entire Parallel operation. Use ContinueOnError to change this.

Slice may only be used with Parallel.

This is a code generation directive.

func Task ¶

func Task(fn interface{}, opts ...TaskOption) Option

Task specifies a task for execution with a Flow or Parallel. A task can be a reference to:

• a top-level function; or
• a bound method; or
• an anonymous function

For example:

// Given,
//   var client *Client
//   func (*Client) GetUser(...) (...)
// The following is a bound method reference.
cff.Task(client.GetUser)

// Given,
//   func bindUser(...) (...)
// The following is a top-level function reference.
cff.Task(bindUser),

// The following is an anonymous function reference.
cff.Task(func(...) (...,, error) {
	// ...
})

A Task's usage and constraints change based on whether you're using it inside a Flow or a Parallel. See the documentation for Flow or Parallel for more details.

This is a code generation directive.

func Tasks ¶

func Tasks(fn ...interface{}) Option

Tasks specifies multiple functions for execution with Parallel. As with Task, each argument to Tasks is a reference to:

• a top-level function; or
• a bound method; or
• an anonymous function

They may all match the signature specified for parallel tasks (see Parallel).

Tasks cannot be used with Flow. Use Task for that.

This is a code generation directive.

func WithEmitter ¶

func WithEmitter(Emitter) Option

WithEmitter provides an optional observer for Flow or Parallel events. Emitters can track metrics, logs, or other observability data.

cff.Flow(ctx,
	// ...
	cff.WithEmitter(em),
)

Provide this option multiple times to use multiple emitters.

WARNING: Do not use this API. We intend to replace it in an upcoming release.

This is a code generation directive.

func NopParallelEmitter ¶

func NopParallelEmitter() ParallelEmitter

NopParallelEmitter is a Parallel emitter that does not do anything.

func SliceEnd ¶

func SliceEnd(fn interface{}) SliceOption

SliceEnd specifies a function for execution when a Slice operation finishes. This function will run after all items in the slice have been processed.

As with parallel tasks, the function passed to SliceEnd may have:

• an optional context.Context parameter
• an optional error return value

Therefore, these are all valid:

cff.SliceEnd(func() {...})
cff.SliceEnd(func() error {...})
cff.SliceEnd(func(ctx context.Context) {...})
cff.SliceEnd(func(ctx context.Context) error {...})

SliceEnd cannot be used with ContinueOnError.

This is a code generation directive.

func NopTaskEmitter ¶

func NopTaskEmitter() TaskEmitter

NopTaskEmitter is a Task emitter that does not do anything.

func FallbackWith ¶

func FallbackWith(results ...interface{}) TaskOption

FallbackWith specifies that if the corresponding task fails with an error or panics, we should recover from that failure and return the provided values instead.

This function accepts the same number of values as returned by the task with exactly the same types -- not including the error return value (if any).

For example:

// Given,
//   func (*Client) ListUsers(context.Context) ([]*User, error)
// And,
//   var cachedUserList []*User = ...
cff.Task(client.ListUsers, cff.FallbackWith(cachedUserList))

If client.ListUsers returns an error or panics, cff will return cachedUserList instead.

This is a code generation directive.

func Instrument ¶

func Instrument(name string) TaskOption

Instrument specifies that this Task should be instrumented for observability. The provided name will be passed to the Emitter you passed into WithEmitter.

This is a code generation directive.

func Invoke ¶

func Invoke(enable bool) TaskOption

Invoke specifies that a flow task must be executed even if none of other tasks consume its output.

By default, flow tasks have the following restrictions:

• must have a non-error return value (outputs)
• the output must be consumed by another task or flow result (via Results)

A task tagged with Invoke(true) loses these restriction. It may have zero outputs, or if it has outputs, other tasks or flow results don't have to consume them.

cff.Task(func(ctx context.Context, req *Request) {
	res, err := shadowClient.Send(req)
	log.Info("shadowed request", "response", res, "error", err)
}, cff.Invoke(true))

This is a code generation directive.

func Predicate ¶

func Predicate(fn interface{}) TaskOption

Predicate specifies a function that determines if the corresponding task should run.

The predicate function has the following signature:

func(I1, I2, ...) bool

Where the arguments I1, I2, ... are inputs similar to a task. Arguments added to the predicate become a dependency of the task, so the predicate or the task will not run until that value is available.

When specified, the corresponding task will be executed only if this function returns true. If the function evaluates to false, the cff will skip execution of this task. If any other tasks depend on this task, cff will give them zero values of the outputs of this task.

For example:

cff.Task(
	authorizeUser,
	cff.Predicate(func(cfg *Config) bool {
		return cfg.Production == true
	}),
)

This is a code generation directive.