README
¶
stream
Install
go get github.com/wushilin/stream
Documentation
$ godoc -http=":16666"
Browse http://localhost:16666
Usage
Stream API
What is a stream? Stream is a concept of collection of items, from various sources.
Creating Stream
There are couple of ways of creating streams. Please read throught code below.
import "github.com/wushilin/stream"
// create stream from static list
s := stream.Of("1","2","3")
// Create stream from slice
s := stream.FromArray([]string{"1","2","3"})
// Create stream from a file
s := stream.FromFileLines("/tmp/input.txt")
// Note that s.Close() must be called for file streams - it will close
// the file handle
// Create stream from a receiver channel
c := make(chan int)
s := stream.FromChannel(c)
// Note here the other go routines must send to the channel and close the channel
// Create stream from a generator function
s := stream.Generate(func() int {
return 5
})
// This is pretty much useless, it returns a infinite stream of integer 5.
// But you can combine it with other features like Limit, Skip, and Sum
s := stream.Generate(func() int {
return rand.Intn(12)
})
// This will create a infinite stream of random integers betwee 0 and 12
// Create stream from iterative function
s := stream.Iterate(1, func(i int) int {
return i + 2
})
// This will generate a stream of even integers, starting from 1.
seed := 1
fibonaci_func := func(i int) int {
result := i + seed
seed = i
return result
}
stream.Iterate(1, fibonaci).Limit(30)
// This creates a stream of Fibonacci numbers, limited to first 30 items
// Create stream from a range
s := stream.Range(1, 100)
fmt.Println(s.Sum())
// You will get 4950, true <- true means there is a sum for retrieval
// Create stream from Iterator
// dummy type implements Iterator, which has a Next() function
// Next() function returns the next value and whether the value exists.
type dummy struct {
}
func (v dummy) Next() (interface{}, bool) {
return 5, true
}
s := stream.FromIterator(dummy{}) // this is same as static generator function
// Create stream from Map keys and values
mp := make(map[string]string)
mp["1"] = "Jessy"
mp["2"] = "Steve"
sk := stream.FromMapKeys(mp)
sv := stream.FromMapValues(mp)
se := stream.FromMapEntries(mp)
Stream operators
s1 := stream.Range(0,5)
s2 := stream.Range(5,10)
s3 := s1.Concat(s2)
// s3 is the same as stream.Range(0,10)
// this applies for all kinds of streams, not only ranges
s1 := stream.Range(0, 5) // this stream, when consumed, will produce 0,1,2,3,4
s2 := s1.Map(func(i int) int {
return i + 5
})
// s2, when consumed, will consume s1, and produce 5,6,7,8,9
s3 := s2.Map(func(i int) string {
return fmt.Sprintf("Student number %d", i)
}
// s3, when consumed, will consume s2 and produce
// "Student number 5", "Student number 6",...,"Student number 9"
The result is a stream.Optional object. The optional object has a Value() function to retrieve its value
s1 := stream.Of(1,2,3,4,5)
sum := 0
reduce_func := func(i,j int) int { return i + j }
fmt.Println(s1.Reduce(reduce_func).Value())
// This will produce 15, true (15 is the result, true means there is a result)
// stream.Range(5, -1).Reduce(reduce_func).Value() will produce nil, false
// since stream.Range(5, -1) contains no elements.
The remaining items of upper stream is not consumed
s1 := stream.Generate(func() int { return 5 }) // s1 is infinite number of 5s
s2 := s1.Limit(20)
fmt.Println(s2.Sum().Value()) // will produce 100, true
fmt.Println(stream.Of("1","2").Count()) // produces 2
fmt.Println(stream.Range(0,100).Count()) // produces 100
fmt.Println(stream.Generate(func() int{ return 5 })) // never returns, infinite loop
s1 := stream.FromFileLines("/tmp/test.txt")
defer s1.Close()
s2 := s1.Filter(func(line string) bool {
return len(line) < 100
})
// s2 is stream of strings, where the lines are shorter than 100 characters
stream.Range(0, 5).Each(func(i int) {
fmt.Println(i, "* 2 =", i*2)
})
// Produces
// 0 * 2 = 0
// 1 * 2 = 2
// ...
// 4 * 2 = 8
Streams by default, has no close handler. so Close() does nothing. The only one has a default close handler is the stream.FromFileLines("filename") When it is closed, the file gets closed. You can attach a close handler, so the handler is called when stream is closed.
s1 := stream.Range(0, 100).OnClose(func() {
fmt.Println("I am closed, good job man, I will call some other functions")
}).OnClose(func() {
fmt.Println("I am closing some database connection")
})
defer s1.Close()
// Then you will see the two OnClose functions are called in order they are attached. (earlier OnClose func got called first).
streams are implemented using iterators. So it simply returns the underlying iterator
iter := stream.Range(0,100).Iterator()
for val, has_next := iter.Next(); val, has_next = iter.Next(); has_next {
// do something
}
s1 := stream.Range(0, 100).Skip(50) // 50, 51, 52,...., 99
buffer := make([]int, 100)
collect_count := stream.Range(0, 100000).CollectTo(buffer)
// collect_count will be 100, and buffer will contain 0~99
collect_count1 := stream.Range(0,1).CollectTo(buffer)
// collect_count1 will be 1, and buffer will contain 0. Rest of values untouched
max_cmp_func := func(i, j int) bool {
return i < j
}
fmt.Println(stream.Range(0, 100).MaxCmp(max_cmp_func)) // 99, true
max_cmp_func := func(i, j int) bool {
return i < j
}
fmt.Println(stream.Range(0, 100).MinCmp(max_cmp_func)) // 99, true
fmt.Println(stream.Range(0, 100).Max().Value()) // 99, true
fmt.Println(stream.Of(1.1, 1.2, 1.3).Min().Value()) // 1.1, true
count := 0
fmt.Println(stream.Range(0, 5).Peek(func(i int) {
count = count + i
}).Count())
fmt.Println(count)
// Produces 5 (the count), 10 (the sum)
stream.WithIndex(stream.Range(6, 100)).Each(func(i stream.NumberedItem) {
fmt.Println("Object index is:", i.Index)
fmt.Println("Object is:", i.Item)
})
destc := make(chan int)
done := make(chan bool)
go func() {
for next_int := range(destc) {
fmt.Println("Received:", next_int)
}
done <- true
}()
stream.Range(0, 100).SendTo(destc)
close(destc)
<- done
// This will send 0~99 (100 numbers) to the channel for another go routine to consume.
Documentation
¶
Overview ¶
Define stream just like java
Index ¶
- Variables
- func CollectAll(iter Iterator) []interface{}
- func CollectN(iter Iterator, count int) []interface{}
- type ChannelIterator
- type GenFunc
- type Iterator
- type LessThanCmpFunc
- type MapCallFunc
- type MapEntry
- type MapEntryIterator
- type MapFunc
- type MapKeyIterator
- type MapValueIterator
- type NumberedItem
- type Optional
- type PredictFunc
- type ProducerFunc
- type ReduceFunc
- type Stream
- func FromArray(it interface{}) Stream
- func FromChannel(it interface{}) Stream
- func FromFileLines(filepath string) (Stream, error)
- func FromIterator(it Iterator) Stream
- func FromMapEntries(it interface{}) Stream
- func FromMapKeys(it interface{}) Stream
- func FromMapValues(it interface{}) Stream
- func FromReader(r io.Reader, delimeter rune) Stream
- func Generate(f interface{}) Stream
- func Iterate(seed interface{}, f interface{}) Stream
- func Of(vars ...interface{}) Stream
- func Range(low, high int) Stream
- func WithIndex(in Stream) Stream
- type ValueFunc
- type VoidFunc
Constants ¶
This section is empty.
Variables ¶
var SumFloat32 = func(i, j interface{}) interface{} { return i.(float32) + j.(float32) }
float32 Sum function. It is also a reduce function
var SumFloat64 = func(i, j interface{}) interface{} { return i.(float64) + j.(float64) }
float64 Sum function. It is also a reduce function
var SumInt = func(i, j interface{}) interface{} { return i.(int) + j.(int) }
Int Sum function. It is also a reduce function
var SumInt32 = func(i, j interface{}) interface{} { return i.(int32) + j.(int32) }
Int32 Sum function. It is also a reduce function
var SumInt64 = func(i, j interface{}) interface{} { return i.(int64) + j.(int64) }
int64 Sum function. It is also a reduce function
var SumUint32 = func(i, j interface{}) interface{} { return i.(uint32) + j.(uint32) }
uint32 Sum function. It is also a reduce function
var SumUint64 = func(i, j interface{}) interface{} { return i.(uint64) + j.(uint64) }
uint64 Sum function. It is also a reduce function
Functions ¶
func CollectN ¶
Collect at most count elements from Iterator object. Returns a slice of object collected. If count is 0, empty slice is returned. If count is less than 0, Maximum number of objects are collected (similar to Integer.MAX_VALUE in java)
Note: Collected objects might be less than count if Iterator reached end Note: Not thread safe
Types ¶
type ChannelIterator ¶
type ChannelIterator struct {
// contains filtered or unexported fields
}
Utility struct to for channel iteration Iterate through channels requires explicit channel closure. Failing to do so will result indefinite waiting.
Note: thread safe
func NewChannelIterator ¶
func NewChannelIterator(target interface{}) *ChannelIterator
Create Iterator on a channel. Target must be a channel
Note: Channel sender need to close channel or Iterator's last Next() call will hang forever Multiple Iterator on same channel is possible since each Next() call will receive from the channel
func (*ChannelIterator) Next ¶
func (v *ChannelIterator) Next() (result interface{}, ok bool)
Implements Iterator. result will be channel's corresponding value type, requires casting
Note: thread safe
type GenFunc ¶
type GenFunc func() interface{}
GenFunc is a function generate values, It is also a ProducerFunc
type Iterator ¶
type Iterator interface {
/*
Next() returns the next element. If an element is available, val is the element and ok is true.
If an element is not available, val is nil, and ok is set as false
A common way of iterating:
for next, ok := iter.Next(); ok; next, ok = iter.Next() {
do_some_thing_with(next)
}
*/
Next() (val interface{}, ok bool)
}
Defined a iterator for generic purpose
func NewArrayIterator ¶
func NewArrayIterator(target interface{}) Iterator
Return a new iterator based on array. target must be an array, or a slice, or a pointer to an array, or a pointer to a slice It can't be pointer to pointer to an array/slice Note: Not thread safe
type LessThanCmpFunc ¶
type LessThanCmpFunc func(x, y interface{}) bool
A LessThanCMpFunc takes two elements, return true if first is smaller than second logically The type of x,y should be the same
var NaturalCmpFunc LessThanCmpFunc = func(i, j interface{}) bool { t1 := reflect.TypeOf(i).Kind() t2 := reflect.TypeOf(j).Kind() if t1 != t2 { panic("Can't compare of different types") } switch t1 { case reflect.Int, reflect.Int16, reflect.Int32, reflect.Int8, reflect.Int64: return reflect.ValueOf(i).Int() < reflect.ValueOf(j).Int() case reflect.Uint, reflect.Uint16, reflect.Uint32, reflect.Uint8, reflect.Uint64: return reflect.ValueOf(i).Uint() < reflect.ValueOf(j).Uint() case reflect.Float32, reflect.Float64: return reflect.ValueOf(i).Float() < reflect.ValueOf(j).Float() case reflect.String: return reflect.ValueOf(i).String() < reflect.ValueOf(i).String() default: panic("Can't compare of non int/float/string variable, please implement yourself") } return false }
Natural LessThanCmp function. Only supports number and strings Panics if trying with other types
type MapCallFunc ¶
type MapCallFunc func(x interface{})
It is like a map function, but produce no result
a -> fmt.Println(a) is a MapCallFunc
type MapEntry ¶
type MapEntry struct {
Key interface{}
Value interface{}
}
Map entry iteration: Just like java.util.Map.Entry class
type MapEntryIterator ¶
type MapEntryIterator struct {
// contains filtered or unexported fields
}
Utility struct for Map Entry iteration. All map keys are obtained when iterator is constructed. Thus no new key's value added after construction will be returned. However, values modified after iteration construction, new key/value will be returned
Note: Not thread safe
func NewMapEntryIterator ¶
func NewMapEntryIterator(target interface{}) *MapEntryIterator
Create Iterator to iterate over a map's entries Target must be a map Multiple Iterator on same map is possible.
func (*MapEntryIterator) Next ¶
func (v *MapEntryIterator) Next() (result interface{}, ok bool)
Implements Iterator. Will return MapEntry on each call.
Note: Not thread safe
type MapFunc ¶
type MapFunc func(x interface{}) interface{}
A map function is a transform function
func(a interface{}) interface{} {
return a
}
is a dummy map function. The type of x and result should be the same
type MapKeyIterator ¶
type MapKeyIterator struct {
// contains filtered or unexported fields
}
Utility struct for map key iteration
Note: Not thread safe
func NewMapKeyIterator ¶
func NewMapKeyIterator(target interface{}) *MapKeyIterator
Create Iterator to iterate over a map's keys. Target must be a map or code may panic Multiple Iterator on same map is possible
func (*MapKeyIterator) Next ¶
func (v *MapKeyIterator) Next() (result interface{}, ok bool)
Implements Iterator. Returns a map key on each Next() call All map keys are obtained when iterator is constructed. Thus no new key added after construction will be returned.
Note: Not thread safe
type MapValueIterator ¶
type MapValueIterator struct {
// contains filtered or unexported fields
}
Utility struct for map value iteration
Note: Not thread safe
func NewMapValueIterator ¶
func NewMapValueIterator(target interface{}) *MapValueIterator
Create Iterator to iterate over a map's values. Target must be a map Multiple Iterator on same map is possible
func (*MapValueIterator) Next ¶
func (v *MapValueIterator) Next() (result interface{}, ok bool)
Implements Iterator. Returns a map value on each Next() call All map keys are obtained when iterator is constructed. Thus no new key's value added after construction will be returned. However, values modified after iteration construction, new value will be returned
Note: Not thread safe
type NumberedItem ¶
type NumberedItem struct {
Index int
Item interface{}
}
type Optional ¶
type Optional interface {
// Return the value and if value doesn't exist, false.
// Caller need to check value exist before the value can be trusted
Value() (interface{}, bool)
// Return another optional value, when call Value() on the new optional,
// return this object's value if present
// If not, return other's Value()
Or(other Optional) Optional
// Return this object's value if present, or the defaultValue supplied
OrValue(defaultV interface{}) interface{}
}
Optional value may or may not have a value
func NewEmptyOptional ¶
func NewEmptyOptional() Optional
Return an optional has no value call to result.Value() will return nil, false
func NewOptional ¶
func NewOptional(val interface{}) Optional
Return an optional has a value call to result.Value() will return val, true
type PredictFunc ¶
type PredictFunc func(x interface{}) bool
Predict function is a filter/test func. Test whether an object match the predict.
a -> a>=5 is a predict function
type ProducerFunc ¶
type ProducerFunc func() interface{}
ProducerFunc is a func that produces a result If your signature doesn't match, you can wrap it with inline func
type ReduceFunc ¶
type ReduceFunc func(x, y interface{}) interface{}
A reduce function is a function that takes 2 element, but return only one. Sum(int, int) int is a reduce function The type of x,y and result should be the same
type Stream ¶
type Stream interface {
// For each element of the stream, apply a map function, and return the
// mapped result stream. It is lazy so only operated when you use a terminal operator
// e.g.
// stream.Range(0, 10).Map(func(a interface{}) interface{} {
// return a.(int) + 1
// }
// will return a lazy stream, from 1~10 (not range(0,10) is 0~9)
Map(f interface{}) Stream
// Use a reduce function to reduce elements in the stream and return reduced resumt.
// e.g.
// stream.Range(0, 10).Reduce(func(a, b interface{}) interface{} {
// if a.(int) > b.(int) {
// return a
// }
// return b
// }
// will return an optional, value is 9. This essentially reduce using Max(a,b) function
// This is a terminal operator
Reduce(f interface{}) Optional
// Lazily limit the elements to process to number
Limit(number int) Stream
// Count elements in stream.
// This is terminal operator
Count() int
// Similar to Map, but skip if the predict func did not pass the test.
// Return a lazy stream
// e.g.
// stream.Range(0, 10).Filer(func(a interface{}) bool) {
// return a.(int) > 5
// }
// will return a stream from [6 ~ 9]
Filter(f interface{}) Stream
// For each of the stream, do the function
// e.g.
// var sum := 0
// stream.Of(1, 2, 3).Each(func(i interface{}) {
// sum = sum + i.(int)
// })
// will return 6
// Note this is a terminal operator
Each(f interface{})
// Close the stream. If your stream if from files, you have to close it.
// Any OnClose handler previously attached will be called
Close()
// Attach another close handler to the stream.
// The handler will be called after the previous handler had been called
OnClose(closefunc func()) Stream
// Return an iterator of elements from the stream
Iterator() Iterator
// Concat the other stream to end of current stream and return the new stream.
// Both original stream and other stream itself are not modified.
Concat(other Stream) Stream
// Collect elements to target. Target must be array/slice of same type as the elements.
// The max number if elements to collect is len(target).
// Returns the number of elements collected
// When number of elements collected equal to slice/array cap, there might be more elemnts to be collected
// otherwise, it is guaranteed no more elements left in stream
CollectTo(target interface{}) int
// Get max in the stream using a less comparator function
MaxCmp(f interface{}) Optional
// Get min in the stream using a less comparator function
MinCmp(f interface{}) Optional
// Get max in the stream, using natural comparator. Only supports numbers and strings
Max() Optional
// Get min in the stream, using natural comparator. Only supports numbers and strings
Min() Optional
// Skip N elements in the stream, returning a new stream
Skip(number int) Stream
// Return a new stream of itself, but when elements are consumed, the peek function is called
// e.g.
// sum := 0
// fmt.Println(stream.Range(0, 10).Peek(func(i interface) {
// sum = sum + i.(int)
// }).Count())
// fmt.Println("Sum is", sum)
// This will count the elements, as well as add a sum
Peek(f interface{}) Stream
// Return sum of the elements, only supports int, int32, int64, uint32, uint64
// float32, float64
// call
// val, ok := result.Value()
// if ok => val can be used
// otherwise, the val can't be used
Sum() Optional
// Stream all elements to channel. May block the caller if
// the channel is full
SendTo(channel interface{})
}
Stream defines java like stream magics
func FromArray ¶
func FromArray(it interface{}) Stream
Return stream from array. It is safe to call multiple FromArray on same array The call doesn't modify source array
func FromChannel ¶
func FromChannel(it interface{}) Stream
Return stream from channel. If you use same channel create multiple streams all streams will share the channel and may see part of the data Sender of channel must close or stream's reduce function/map function may not terminate
func FromFileLines ¶
Return a stream from file lines. Caller must call close on return stream. If file open fails, the stream will be nil and error will be returned.
func FromIterator ¶
Return stream from iterator. stream's reduce function will consume all items in iterator
func FromMapEntries ¶
func FromMapEntries(it interface{}) Stream
Create stream from map's key value pairs. Note the iterator is a snapshot of map subsequent modification after Stream is created won't be visiable to stream
func FromMapKeys ¶
func FromMapKeys(it interface{}) Stream
Create stream from map's keys. Note the iterator is a snapshot of map subsequent modification after Stream is created won't be visiable to stream
func FromMapValues ¶
func FromMapValues(it interface{}) Stream
Create stream from map's values. Note the iterator is a snapshot of map subsequent modification after Stream is created won't be visiable to stream
func FromReader ¶
Create new stream from a reader, with the delimiter for tokenizer
func Generate ¶
func Generate(f interface{}) Stream
Generate will use the GenFunc to generate a infinite stream e.g.
stream.Generate(func() interface{} {
return 5
}
will generate a infinite stream of 5. Note it is lazy so do not count infinite stream, it will not complete Similarly, do not Reduce infinite stream You can limit first before reducing
func Iterate ¶
func Iterate(seed interface{}, f interface{}) Stream
Generate a infinite stream, using seed as first element, then use MapFun and the previously returned value to generate the new value e.g.
func add1(i interface{}) interface{} {
return i.(int) + 1
}
stream.Iterate(1, add1).Limit(3) <= will produce the same as
stream.Of(1, 2, 3)
func Of ¶
func Of(vars ...interface{}) Stream
Create a stream from a set of values. e.g.
stream.Of(1,2,3,4,5).Sum().Value() => 15, true
15 is the result, true means the optional actually have a value
stream.Of().Sum().Value() => nil, false
nil is default result, false means the stream is empty, so sum is non-existent
Source Files