fx

Documentation

Index

• Variables
• func DoWithRetries(fn func() error, opts ...RetryOption) error
• func DoWithTimeout(fn func() error, timeout time.Duration, opts ...FxOption) error
• func Parallel(fns ...func())
• type FilterFunc
• type ForAllFunc
• type ForEachFunc
• type FxOption
  • func WithContext(ctx context.Context) FxOption
• type GenerateFunc
• type KeyFunc
• type LessFunc
• type MapFunc
• type Option
  • func UnlimitedWorkers() Option
  • func WithWorkers(workers int) Option
• type ParallelFunc
• type ReduceFunc
• type RetryOption
  • func WithRetries(times int) RetryOption
• type Stream
  • func From(generate GenerateFunc) Stream
  • func Just(items ...interface{}) Stream
  • func Range(source <-chan interface{}) Stream
  • func (p Stream) Buffer(n int) Stream
  • func (p Stream) Count() (count int)
  • func (p Stream) Distinct(fn KeyFunc) Stream
  • func (p Stream) Done()
  • func (p Stream) Filter(fn FilterFunc, opts ...Option) Stream
  • func (p Stream) ForAll(fn ForAllFunc)
  • func (p Stream) ForEach(fn ForEachFunc)
  • func (p Stream) Group(fn KeyFunc) Stream
  • func (p Stream) Head(n int64) Stream
  • func (p Stream) Map(fn MapFunc, opts ...Option) Stream
  • func (p Stream) Merge() Stream
  • func (p Stream) Parallel(fn ParallelFunc, opts ...Option)
  • func (p Stream) Reduce(fn ReduceFunc) (interface{}, error)
  • func (p Stream) Reverse() Stream
  • func (p Stream) Sort(less LessFunc) Stream
  • func (p Stream) Split(n int) Stream
  • func (p Stream) Tail(n int64) Stream
  • func (p Stream) Walk(fn WalkFunc, opts ...Option) Stream
• type WalkFunc

Variables

var (
	ErrCanceled = context.Canceled
	ErrTimeout  = context.DeadlineExceeded
)

Functions

func DoWithRetries

func DoWithRetries(fn func() error, opts ...RetryOption) error

func DoWithTimeout

func DoWithTimeout(fn func() error, timeout time.Duration, opts ...FxOption) error

func Parallel

func Parallel(fns ...func())

Types

type FilterFunc

type FilterFunc func(item interface{}) bool

type ForAllFunc

type ForAllFunc func(pipe <-chan interface{})

type ForEachFunc

type ForEachFunc func(item interface{})

type FxOption

type FxOption func() context.Context

func WithContext

func WithContext(ctx context.Context) FxOption

type GenerateFunc

type GenerateFunc func(source chan<- interface{})

type KeyFunc

type KeyFunc func(item interface{}) interface{}

type LessFunc

type LessFunc func(a, b interface{}) bool

type MapFunc

type MapFunc func(item interface{}) interface{}

type Option

type Option func(opts *rxOptions)

func UnlimitedWorkers

func UnlimitedWorkers() Option

UnlimitedWorkers lets the caller to use as many workers as the tasks.

func WithWorkers

func WithWorkers(workers int) Option

WithWorkers lets the caller to customize the concurrent workers.

type ParallelFunc

type ParallelFunc func(item interface{})

type ReduceFunc

type ReduceFunc func(pipe <-chan interface{}) (interface{}, error)

type RetryOption

type RetryOption func(*retryOptions)

func WithRetries

func WithRetries(times int) RetryOption

type Stream

type Stream struct {
	// contains filtered or unexported fields
}

func From

func From(generate GenerateFunc) Stream

From constructs a Stream from the given GenerateFunc.

func Just

func Just(items ...interface{}) Stream

Just converts the given arbitrary items to a Stream.

func Range

func Range(source <-chan interface{}) Stream

Range converts the given channel to a Stream.

func (Stream) Buffer

func (p Stream) Buffer(n int) Stream

Buffer buffers the items into a queue with size n. It can balance the producer and the consumer if their processing throughput don't match.

func (Stream) Count

func (p Stream) Count() (count int)

Count counts the number of elements in the result.

func (Stream) Distinct

func (p Stream) Distinct(fn KeyFunc) Stream

Distinct removes the duplicated items base on the given KeyFunc.

func (Stream) Done

func (p Stream) Done()

Done waits all upstreaming operations to be done.

func (Stream) Filter

func (p Stream) Filter(fn FilterFunc, opts ...Option) Stream

Filter filters the items by the given FilterFunc.

func (Stream) ForAll

func (p Stream) ForAll(fn ForAllFunc)

ForAll handles the streaming elements from the source and no later streams.

func (Stream) ForEach

func (p Stream) ForEach(fn ForEachFunc)

ForEach seals the Stream with the ForEachFunc on each item, no successive operations.

func (Stream) Group

func (p Stream) Group(fn KeyFunc) Stream

Group groups the elements into different groups based on their keys.

func (Stream) Head

func (p Stream) Head(n int64) Stream

func (Stream) Map

func (p Stream) Map(fn MapFunc, opts ...Option) Stream

Maps converts each item to another corresponding item, which means it's a 1:1 model.

func (Stream) Merge

func (p Stream) Merge() Stream

Merge merges all the items into a slice and generates a new stream.

func (Stream) Parallel

func (p Stream) Parallel(fn ParallelFunc, opts ...Option)

Parallel applies the given ParallelFunc to each item concurrently with given number of workers.

func (Stream) Reduce

func (p Stream) Reduce(fn ReduceFunc) (interface{}, error)

Reduce is a utility method to let the caller deal with the underlying channel.

func (Stream) Reverse

func (p Stream) Reverse() Stream

Reverse reverses the elements in the stream.

func (Stream) Sort

func (p Stream) Sort(less LessFunc) Stream

Sort sorts the items from the underlying source.

func (Stream) Split

func (p Stream) Split(n int) Stream

Split splits the elements into chunk with size up to n, might be less than n on tailing elements.

func (Stream) Tail

func (p Stream) Tail(n int64) Stream

func (Stream) Walk

func (p Stream) Walk(fn WalkFunc, opts ...Option) Stream

Walk lets the callers handle each item, the caller may write zero, one or more items base on the given item.

type WalkFunc

type WalkFunc func(item interface{}, pipe chan<- interface{})