stream

README

stream

流式计算工具包 (Streaming Computing SDK)

TODO

Documentation

Index

• Variables
• func AnyTo[T any](data ...any) types.Collector[any]
• func To[T, R any](converter types.Converter[T, R]) types.Collector[T]
• type Sortable
  • func (a *Sortable[T]) Len() int
  • func (a *Sortable[T]) Less(i, j int) bool
  • func (a *Sortable[T]) Swap(i, j int)
• type Streamer
  • func Concat[T any](dst Streamer[T], srcs ...Streamer[T]) Streamer[T]
  • func Of[T any](supply types.Supplier[T]) Streamer[T]
  • func Repeat[T any](t T) Streamer[T]
  • func RepeatN[T any](t T, count int64) Streamer[T]
  • func SliceOf[T any](slice ...T) Streamer[T]

Variables

var (
	// ErrUnsupportType unsupport type
	ErrUnsupportType = errors.New("unsupport type")
)

Functions

func AnyTo

func AnyTo[T any](data ...any) types.Collector[any]

AnyTo converts a slice of any to a slice of T

func To

func To[T, R any](converter types.Converter[T, R]) types.Collector[T]

To converts a slice of T to a slice of R

Types

type Sortable

type Sortable[T any] struct {
	List []T
	Cmp  types.Comparator[T]
}

Sortable implement sort.Interface

func (*Sortable[T]) Len

func (a *Sortable[T]) Len() int

implement sort.Interface

func (*Sortable[T]) Less

func (a *Sortable[T]) Less(i, j int) bool

func (*Sortable[T]) Swap

func (a *Sortable[T]) Swap(i, j int)

type Streamer

type Streamer[T any] interface {
	// WithContext set Streamer context
	WithContext(context.Context) Streamer[T]

	// Filter filter data by Judge result
	Filter(types.Judge[T]) Streamer[T]
	Map(types.Mapper[T]) Streamer[T]
	Convert(types.Converter[T, any]) Streamer[any]
	Peek(types.Consumer[T]) Streamer[T]

	Distinct() Streamer[T]
	Sort(types.Comparator[T]) Streamer[T]
	ReverseSort(types.Comparator[T]) Streamer[T]
	Reverse() Streamer[T]
	Limit(int64) Streamer[T]
	Skip(int64) Streamer[T]
	Pick(startIndex, endIndex, interval int) Streamer[T]

	// Append append data to streamer source
	Append(...T) Streamer[T]
	// Execute eager execute streamer stage
	Execute() Streamer[T]

	// Parallel 0 do nothing, 1 async work, 2-n concurrent work
	Parallel(int) Streamer[T]

	// ToSlice
	ToSlice() []T
	Collect(types.Collector[T]) any
	// ForEach
	ForEach(types.Consumer[T])
	// Match methods
	AllMatch(types.Judge[T]) bool
	NonMatch(types.Judge[T]) bool
	AnyMatch(types.Judge[T]) bool
	// Reduce reduce calculate
	Reduce(accumulator types.BinaryOperator[T]) T
	ReduceFrom(initValue T, accumulator types.BinaryOperator[T]) T
	ReduceWith(initValue any, accumulator types.Accumulator[T, any]) any
	// ReduceBy use `buildInitValue` to build the initValue, which parameter is a int64 means element size, or -1 if unknown size.
	// Then use `accumulator` to add each element to previous result
	ReduceBy(initValueBulider func(sizeMayNegative int) any, accumulator types.Accumulator[T, any]) any
	// Pick one
	First() T
	Take() T
	Any() T
	Last() T
	// Cout return count result
	Count() int64
}

func Concat

func Concat[T any](dst Streamer[T], srcs ...Streamer[T]) Streamer[T]

Concat concat streamers

func Of

func Of[T any](supply types.Supplier[T]) Streamer[T]

Of create a new stream with supply

func Repeat

func Repeat[T any](t T) Streamer[T]

Repeat create a new stream with unlimit repeated data items

func RepeatN

func RepeatN[T any](t T, count int64) Streamer[T]

RepeatN create a new stream with n times repeated data items

func SliceOf

func SliceOf[T any](slice ...T) Streamer[T]

SliceOf receive array and initlize streamer