xstream

Documentation

Index

• Variables
• type Collector
• type CollectorFunc
  • func (cf CollectorFunc) Input(c <-chan interface{})
• type FilterFunc
• type ForAllFunc
• type ForEachFunc
• type GenerateFunc
• type Group
  • func GroupBy(f KeyFunc) *Group
  • func (g *Group) Input(c <-chan interface{})
  • func (g *Group) Map() map[interface{}][]interface{}
• type KeyFunc
• type LessFunc
• type MapFunc
• type Option
  • func WithOption(options *Options) Option
  • func WithWorkSize(size int) Option
• type Options
• type ParallelFunc
• type ReduceFunc
• type Stream
  • func Concat(a *Stream, others ...*Stream) *Stream
  • func Empty() *Stream
  • func From(generate GenerateFunc) *Stream
  • func Of(items ...interface{}) *Stream
  • func Range(source <-chan interface{}) *Stream
  • func (s *Stream) AllMach(f func(item interface{}) bool) (isFind bool)
  • func (s *Stream) AnyMach(f func(item interface{}) bool) (isFind bool)
  • func (s *Stream) Buffer(n int) *Stream
  • func (s *Stream) Chan() <-chan interface{}
  • func (s *Stream) Collection(collector Collector)
  • func (s *Stream) Concat(others ...*Stream) *Stream
  • func (s *Stream) Copy(streamParam map[string]int) (streamMap map[string]*Stream)
  • func (s *Stream) Count() (count int)
  • func (s *Stream) Distinct(f KeyFunc) *Stream
  • func (s *Stream) Done()
  • func (s *Stream) Filter(fn FilterFunc, opts ...Option) *Stream
  • func (s *Stream) FindFirst() (result interface{}, err error)
  • func (s *Stream) FindLast() (result interface{}, err error)
  • func (s *Stream) FlatMap(fn MapFunc, opts ...Option) *Stream
  • func (s *Stream) Foreach(f ForEachFunc)
  • func (s *Stream) ForeachOrdered(f ForEachFunc)
  • func (s *Stream) Group(f KeyFunc) *Stream
  • func (s *Stream) Limit(size int) *Stream
  • func (s *Stream) Map(fn MapFunc, opts ...Option) *Stream
  • func (s *Stream) Merge() *Stream
  • func (s *Stream) ParallelFinish(fn ParallelFunc, opts ...Option)
  • func (s *Stream) Peek(f ForEachFunc) *Stream
  • func (s *Stream) Reverse() *Stream
  • func (s *Stream) Skip(size int) *Stream
  • func (s *Stream) Sort(less LessFunc) *Stream
  • func (s *Stream) Split(n int) *Stream
  • func (s *Stream) SplitSteam(n int) *Stream
  • func (s *Stream) Tail(n int) *Stream
  • func (s *Stream) Walk(f WalkFunc, opts ...Option) *Stream
• type WalkFunc

Variables

var (

	// ErrNoElement no element error.
	ErrNoElement = errors.New("no element")
)

Types

type Collector

type Collector interface {
	Input(c <-chan interface{})
}

Collector represents a stream collector to collect items

type CollectorFunc

type CollectorFunc func(c <-chan interface{})

CollectorFunc represents a stream collector.

func (CollectorFunc) Input

func (cf CollectorFunc) Input(c <-chan interface{})

Input implements Collector.

type FilterFunc

type FilterFunc func(item interface{}) bool

FilterFunc defines the method to filter a Stream.

type ForAllFunc

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

ForAllFunc defines the method to handle all elements in a Stream.

type ForEachFunc

type ForEachFunc func(item interface{})

ForEachFunc defines the method to handle each element in a Stream.

type GenerateFunc

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

GenerateFunc defines the method to send elements into a Stream.

type Group

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

Group represents a group collector.

func GroupBy

func GroupBy(f KeyFunc) *Group

GroupBy returns a Group.

func (*Group) Input

func (g *Group) Input(c <-chan interface{})

Input implements Collector.

func (*Group) Map

func (g *Group) Map() map[interface{}][]interface{}

Map returns a map.

type KeyFunc

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

KeyFunc defines the method to generate keys for the elements in a Stream.

type LessFunc

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

LessFunc defines the method to compare the elements in a Stream.

type MapFunc

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

MapFunc defines the method to map each element to another object in a Stream.

type Option

type Option func(options *Options)

Option defines the method to customize a Stream.

func WithOption

func WithOption(options *Options) Option

WithOption return a Option interface

func WithWorkSize

func WithWorkSize(size int) Option

WithWorkSize return a Option that set size of work

type Options

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

Options defines the struct to customize a Stream.

type ParallelFunc

type ParallelFunc func(item interface{})

ParallelFunc defines the method to handle elements parallelly.

type ReduceFunc

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

ReduceFunc defines the method to reduce all the elements in a Stream.

type Stream

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

Stream Represents a stream.

func Concat

func Concat(a *Stream, others ...*Stream) *Stream

Concat Returns a concat Stream.

func Empty

func Empty() *Stream

Empty Returns an empty stream.

func From

func From(generate GenerateFunc) *Stream

From Returns a Stream from generate function.

func Of

func Of(items ...interface{}) *Stream

Of Returns a Stream based any element

func Range

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

Range Returns a Stream from source channel.

func (*Stream) AllMach

func (s *Stream) AllMach(f func(item interface{}) bool) (isFind bool)

AllMach Returns whether all elements of this stream match the provided predicate. May not evaluate the predicate on all elements if not necessary for determining the result. If the stream is empty then true is returned and the predicate is not evaluated.

func (*Stream) AnyMach

func (s *Stream) AnyMach(f func(item interface{}) bool) (isFind bool)

AnyMach Returns whether any elements of this stream match the provided predicate. May not evaluate the predicate on all elements if not necessary for determining the result. If the stream is empty then false is returned and the predicate is not evaluated.

func (*Stream) Buffer

func (s *Stream) Buffer(n int) *Stream

Buffer Returns a buffer Stream.

func (*Stream) Chan

func (s *Stream) Chan() <-chan interface{}

Chan Returns a channel of Stream.

func (*Stream) Collection

func (s *Stream) Collection(collector Collector)

Collection collects a Stream.

func (*Stream) Concat

func (s *Stream) Concat(others ...*Stream) *Stream

Concat Returns a Stream that concat others streams

func (*Stream) Copy

func (s *Stream) Copy(streamParam map[string]int) (streamMap map[string]*Stream)

Copy returns multiple streams copied. streamParam specifies the name and buffer size of the replicated stream.

func (*Stream) Count

func (s *Stream) Count() (count int)

Count Returns a number that the elements total size.

func (*Stream) Distinct

func (s *Stream) Distinct(f KeyFunc) *Stream

Distinct Returns a distinct Stream.

func (*Stream) Done

func (s *Stream) Done()

Done Stream.

func (*Stream) Filter

func (s *Stream) Filter(fn FilterFunc, opts ...Option) *Stream

Filter Returns a Stream that

func (*Stream) FindFirst

func (s *Stream) FindFirst() (result interface{}, err error)

FindFirst Returns an interface{} the first element of this stream, or a nil and a error if the stream is empty. If the stream has no encounter order, then any element may be returned

func (*Stream) FindLast

func (s *Stream) FindLast() (result interface{}, err error)

FindLast Returns an interface{} the last element of this stream, or a nil and a error if the stream is empty. If the stream has no encounter order, then any element may be returned

func (*Stream) FlatMap

func (s *Stream) FlatMap(fn MapFunc, opts ...Option) *Stream

FlatMap Returns a Stream consisting of the results of replacing each element of this stream with the contents of a mapped stream produced by applying the provided mapping function to each element. Each mapped stream is closed after its contents have been placed into this stream. (If a mapped stream is null an empty stream is used, instead.

func (*Stream) Foreach

func (s *Stream) Foreach(f ForEachFunc)

Foreach Traversals all elements.

func (*Stream) ForeachOrdered

func (s *Stream) ForeachOrdered(f ForEachFunc)

ForeachOrdered Traversals all elements in reverse order.

func (*Stream) Group

func (s *Stream) Group(f KeyFunc) *Stream

Group Returns a Stream that groups the elements into different groups based on their keys.

func (*Stream) Limit

func (s *Stream) Limit(size int) *Stream

Limit Returns a Stream that contains size elements.

func (*Stream) Map

func (s *Stream) Map(fn MapFunc, opts ...Option) *Stream

Map Returns a Stream consisting of the results of applying the given function to the elements of this stream.

func (*Stream) Merge

func (s *Stream) Merge() *Stream

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

func (*Stream) ParallelFinish

func (s *Stream) ParallelFinish(fn ParallelFunc, opts ...Option)

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

func (*Stream) Peek

func (s *Stream) Peek(f ForEachFunc) *Stream

Peek Returns a Stream consisting of the elements of this stream, additionally performing the provided action on each element as elements are consumed from the resulting stream.

func (*Stream) Reverse

func (s *Stream) Reverse() *Stream

Reverse Returns a Stream that reverses the elements.

func (*Stream) Skip

func (s *Stream) Skip(size int) *Stream

Skip Returns a Stream that skips size elements.

func (*Stream) Sort

func (s *Stream) Sort(less LessFunc) *Stream

Sort Returns a sorted Stream.

func (*Stream) Split

func (s *Stream) Split(n int) *Stream

Split Returns a split Stream that contains multiple slices of chunk size n.

func (*Stream) SplitSteam

func (s *Stream) SplitSteam(n int) *Stream

SplitSteam Returns a split Stream that contains multiple stream of chunk size n.

func (*Stream) Tail

func (s *Stream) Tail(n int) *Stream

Tail Returns a Stream that has n element at the end.

func (*Stream) Walk

func (s *Stream) Walk(f WalkFunc, opts ...Option) *Stream

Walk Returns a Stream that 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{})

WalkFunc defines the method to walk through all the elements in a Stream.