fx

README

fx

a function extension package, with synchronous and asynchronous support.

Using type parameters as method receiver is unsupported, liking func [T] (T) Xxx(...), and type's method must have no type parameters, so i can only use the fucking interface{}.

Usage

map and reduce

  sum, err := From(Range(1, 100)).
    Map(func(v Any) (Any, error) {
      return v.(int) * 2, nil
    }).
    Reduce(0, func(sum, v Any) (Any, error) {
      return sum.(int) + v.(int), nil
    })

async and spawn

  list, err := From(Infinite()).
    Map(func(v Any) (Any, error) {
      return v.(uint64) * v.(uint64), nil
    }).
    Async(10).                        // async the previous map iter, and the size of chan buf is 10
    Spawn(10, func(s Stream) Stream { // spawn 10 goroutines to cousume the iter
      return s.
        Map(func(v Any) (Any, error) {
          time.Sleep(time.Millisecond * 100)
          return v, nil
        }).
        Filter(func(v Any) (bool, error) {
          return v.(uint64)%10 == 4, nil
        }).
        FlatMap(func(v Any) ([]Any, error) {
          iv := v.(uint64)
          return []Any{iv, iv + 1, iv + 2}, nil
        })
    }).
    OnError(func(error) error {
      // log...
      return nil // skip the err
    }).
    Take(100)

• async: make the previous iter execute asynchronously
• spawn: dispatch to multiple groutines and fan-in

Documentation

Index

• func IsNone(err error) bool
• func None() error
• type Any
• type Collect
• type ErrorFunc
• type FilterFunc
• type FlatMapFunc
• type ForEachFunc
• type Fx
  • func From(it Iterator) Fx
  • func FromCompIter(it interface{}) Fx
  • func FromFn(fn interface{}) Fx
• type Iterator
  • func AdaptIter(v interface{}) Iterator
  • func Infinite() Iterator
  • func Range(from, to int) Iterator
  • func WrapFn(fn interface{}) Iterator
• type MapFunc
• type RecoverFunc
• type ReduceFunc
• type SpawnFunc
• type Stream

Functions

func IsNone

func IsNone(err error) bool

func None

func None() error

Types

type Any

type Any interface{}

type Collect

type Collect interface {
	ForEach(fn ForEachFunc) error
	Reduce(sum Any, fn ReduceFunc) (Any, error)
	List(initCap uint) ([]Any, error)
	Take(n uint) ([]Any, error)
}

type ErrorFunc

type ErrorFunc func(error) error

type FilterFunc

type FilterFunc func(Any) (bool, error)

type FlatMapFunc

type FlatMapFunc func(Any) ([]Any, error)

type ForEachFunc

type ForEachFunc func(Any) (bool, error)

type Fx

type Fx interface {
	Iterator
	Stream
	Collect
}

func From

func From(it Iterator) Fx

func FromCompIter

func FromCompIter(it interface{}) Fx

func FromFn

func FromFn(fn interface{}) Fx

type Iterator

type Iterator interface {
	Next() (Any, error)
	Close()
}

func AdaptIter

func AdaptIter(v interface{}) Iterator

func Infinite

func Infinite() Iterator

func Range

func Range(from, to int) Iterator

func WrapFn

func WrapFn(fn interface{}) Iterator

type MapFunc

type MapFunc func(Any) (Any, error)

type RecoverFunc

type RecoverFunc func(p interface{}) error

type ReduceFunc

type ReduceFunc func(sum Any, v Any) (Any, error)

type SpawnFunc

type SpawnFunc func(Stream) Stream

type Stream

type Stream interface {
	Map(MapFunc) Fx
	FlatMap(fn FlatMapFunc) Fx
	Filter(fn FilterFunc) Fx
	Async(bs uint) Fx
	Spawn(n uint, fn SpawnFunc) Fx
	OnError(fn ErrorFunc) Fx
	Recover(fn RecoverFunc) Fx
	// contains filtered or unexported methods
}