feat: implement fx.NoneMatch, fx.First, fx.Last (#1402)
* chore: use workers from options in fx.unlimitedWalk * feat: add fx.NoneMatch * feat: add fx.First, fx.Last * chore: add more comments * docs: add mr readme
This commit is contained in:
Kevin Wan
GitHub
@@ -90,6 +90,7 @@ func Range(source <-chan interface{}) Stream {
|
||||
func (s Stream) AllMach(predicate func(item interface{}) bool) bool {
|
||||
for item := range s.source {
|
||||
if !predicate(item) {
|
||||
drain(s.source)
|
||||
return false
|
||||
}
|
||||
}
|
||||
@@ -103,6 +104,7 @@ func (s Stream) AllMach(predicate func(item interface{}) bool) bool {
|
||||
func (s Stream) AnyMach(predicate func(item interface{}) bool) bool {
|
||||
for item := range s.source {
|
||||
if predicate(item) {
|
||||
drain(s.source)
|
||||
return true
|
||||
}
|
||||
}
|
||||
@@ -186,8 +188,7 @@ func (s Stream) Distinct(fn KeyFunc) Stream {
|
||||
|
||||
// Done waits all upstreaming operations to be done.
|
||||
func (s Stream) Done() {
|
||||
for range s.source {
|
||||
}
|
||||
drain(s.source)
|
||||
}
|
||||
|
||||
// Filter filters the items by the given FilterFunc.
|
||||
@@ -199,9 +200,22 @@ func (s Stream) Filter(fn FilterFunc, opts ...Option) Stream {
|
||||
}, opts...)
|
||||
}
|
||||
|
||||
// First returns the first item, nil if no items.
|
||||
func (s Stream) First() interface{} {
|
||||
for item := range s.source {
|
||||
// make sure the former goroutine not block, and current func returns fast.
|
||||
go drain(s.source)
|
||||
return item
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ForAll handles the streaming elements from the source and no later streams.
|
||||
func (s Stream) ForAll(fn ForAllFunc) {
|
||||
fn(s.source)
|
||||
// avoid goroutine leak on fn not consuming all items.
|
||||
drain(s.source)
|
||||
}
|
||||
|
||||
// ForEach seals the Stream with the ForEachFunc on each item, no successive operations.
|
||||
@@ -246,11 +260,14 @@ func (s Stream) Head(n int64) Stream {
|
||||
}
|
||||
if n == 0 {
|
||||
// let successive method go ASAP even we have more items to skip
|
||||
// why we don't just break the loop, because if breaks,
|
||||
// this former goroutine will block forever, which will cause goroutine leak.
|
||||
close(source)
|
||||
// why we don't just break the loop, and drain to consume all items.
|
||||
// because if breaks, this former goroutine will block forever,
|
||||
// which will cause goroutine leak.
|
||||
drain(s.source)
|
||||
}
|
||||
}
|
||||
// not enough items in s.source, but we need to let successive method to go ASAP.
|
||||
if n > 0 {
|
||||
close(source)
|
||||
}
|
||||
@@ -259,6 +276,13 @@ func (s Stream) Head(n int64) Stream {
|
||||
return Range(source)
|
||||
}
|
||||
|
||||
// Last returns the last item, or nil if no items.
|
||||
func (s Stream) Last() (item interface{}) {
|
||||
for item = range s.source {
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Map converts each item to another corresponding item, which means it's a 1:1 model.
|
||||
func (s Stream) Map(fn MapFunc, opts ...Option) Stream {
|
||||
return s.Walk(func(item interface{}, pipe chan<- interface{}) {
|
||||
@@ -280,6 +304,20 @@ func (s Stream) Merge() Stream {
|
||||
return Range(source)
|
||||
}
|
||||
|
||||
// NoneMatch returns whether all elements of this stream don't 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 (s Stream) NoneMatch(predicate func(item interface{}) bool) bool {
|
||||
for item := range s.source {
|
||||
if predicate(item) {
|
||||
drain(s.source)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// Parallel applies the given ParallelFunc to each item concurrently with given number of workers.
|
||||
func (s Stream) Parallel(fn ParallelFunc, opts ...Option) {
|
||||
s.Walk(func(item interface{}, pipe chan<- interface{}) {
|
||||
@@ -411,15 +449,12 @@ func (s Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
|
||||
var wg sync.WaitGroup
|
||||
pool := make(chan lang.PlaceholderType, option.workers)
|
||||
|
||||
for {
|
||||
for item := range s.source {
|
||||
// important, used in another goroutine
|
||||
val := item
|
||||
pool <- lang.Placeholder
|
||||
item, ok := <-s.source
|
||||
if !ok {
|
||||
<-pool
|
||||
break
|
||||
}
|
||||
|
||||
wg.Add(1)
|
||||
|
||||
// better to safely run caller defined method
|
||||
threading.GoSafe(func() {
|
||||
defer func() {
|
||||
@@ -427,7 +462,7 @@ func (s Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
|
||||
<-pool
|
||||
}()
|
||||
|
||||
fn(item, pipe)
|
||||
fn(val, pipe)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -439,22 +474,19 @@ func (s Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
|
||||
}
|
||||
|
||||
func (s Stream) walkUnlimited(fn WalkFunc, option *rxOptions) Stream {
|
||||
pipe := make(chan interface{}, defaultWorkers)
|
||||
pipe := make(chan interface{}, option.workers)
|
||||
|
||||
go func() {
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for {
|
||||
item, ok := <-s.source
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
|
||||
for item := range s.source {
|
||||
// important, used in another goroutine
|
||||
val := item
|
||||
wg.Add(1)
|
||||
// better to safely run caller defined method
|
||||
threading.GoSafe(func() {
|
||||
defer wg.Done()
|
||||
fn(item, pipe)
|
||||
fn(val, pipe)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -465,14 +497,14 @@ func (s Stream) walkUnlimited(fn WalkFunc, option *rxOptions) Stream {
|
||||
return Range(pipe)
|
||||
}
|
||||
|
||||
// UnlimitedWorkers lets the caller to use as many workers as the tasks.
|
||||
// UnlimitedWorkers lets the caller use as many workers as the tasks.
|
||||
func UnlimitedWorkers() Option {
|
||||
return func(opts *rxOptions) {
|
||||
opts.unlimitedWorkers = true
|
||||
}
|
||||
}
|
||||
|
||||
// WithWorkers lets the caller to customize the concurrent workers.
|
||||
// WithWorkers lets the caller customize the concurrent workers.
|
||||
func WithWorkers(workers int) Option {
|
||||
return func(opts *rxOptions) {
|
||||
if workers < minWorkers {
|
||||
@@ -483,6 +515,7 @@ func WithWorkers(workers int) Option {
|
||||
}
|
||||
}
|
||||
|
||||
// buildOptions returns a rxOptions with given customizations.
|
||||
func buildOptions(opts ...Option) *rxOptions {
|
||||
options := newOptions()
|
||||
for _, opt := range opts {
|
||||
@@ -492,6 +525,13 @@ func buildOptions(opts ...Option) *rxOptions {
|
||||
return options
|
||||
}
|
||||
|
||||
// drain drains the given channel.
|
||||
func drain(channel <-chan interface{}) {
|
||||
for range channel {
|
||||
}
|
||||
}
|
||||
|
||||
// newOptions returns a default rxOptions.
|
||||
func newOptions() *rxOptions {
|
||||
return &rxOptions{
|
||||
workers: defaultWorkers,
|
||||
|
||||
@@ -17,320 +17,489 @@ import (
|
||||
)
|
||||
|
||||
func TestBuffer(t *testing.T) {
|
||||
const N = 5
|
||||
var count int32
|
||||
var wait sync.WaitGroup
|
||||
wait.Add(1)
|
||||
From(func(source chan<- interface{}) {
|
||||
ticker := time.NewTicker(10 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
const N = 5
|
||||
var count int32
|
||||
var wait sync.WaitGroup
|
||||
wait.Add(1)
|
||||
From(func(source chan<- interface{}) {
|
||||
ticker := time.NewTicker(10 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
|
||||
for i := 0; i < 2*N; i++ {
|
||||
select {
|
||||
case source <- i:
|
||||
atomic.AddInt32(&count, 1)
|
||||
case <-ticker.C:
|
||||
wait.Done()
|
||||
return
|
||||
for i := 0; i < 2*N; i++ {
|
||||
select {
|
||||
case source <- i:
|
||||
atomic.AddInt32(&count, 1)
|
||||
case <-ticker.C:
|
||||
wait.Done()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}).Buffer(N).ForAll(func(pipe <-chan interface{}) {
|
||||
wait.Wait()
|
||||
// why N+1, because take one more to wait for sending into the channel
|
||||
assert.Equal(t, int32(N+1), atomic.LoadInt32(&count))
|
||||
}).Buffer(N).ForAll(func(pipe <-chan interface{}) {
|
||||
wait.Wait()
|
||||
// why N+1, because take one more to wait for sending into the channel
|
||||
assert.Equal(t, int32(N+1), atomic.LoadInt32(&count))
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func TestBufferNegative(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Buffer(-1).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Buffer(-1).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
assert.Equal(t, 10, result)
|
||||
})
|
||||
assert.Equal(t, 10, result)
|
||||
}
|
||||
|
||||
func TestCount(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
elements []interface{}
|
||||
}{
|
||||
{
|
||||
name: "no elements with nil",
|
||||
},
|
||||
{
|
||||
name: "no elements",
|
||||
elements: []interface{}{},
|
||||
},
|
||||
{
|
||||
name: "1 element",
|
||||
elements: []interface{}{1},
|
||||
},
|
||||
{
|
||||
name: "multiple elements",
|
||||
elements: []interface{}{1, 2, 3},
|
||||
},
|
||||
}
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
elements []interface{}
|
||||
}{
|
||||
{
|
||||
name: "no elements with nil",
|
||||
},
|
||||
{
|
||||
name: "no elements",
|
||||
elements: []interface{}{},
|
||||
},
|
||||
{
|
||||
name: "1 element",
|
||||
elements: []interface{}{1},
|
||||
},
|
||||
{
|
||||
name: "multiple elements",
|
||||
elements: []interface{}{1, 2, 3},
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
val := Just(test.elements...).Count()
|
||||
assert.Equal(t, len(test.elements), val)
|
||||
})
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
val := Just(test.elements...).Count()
|
||||
assert.Equal(t, len(test.elements), val)
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestDone(t *testing.T) {
|
||||
var count int32
|
||||
Just(1, 2, 3).Walk(func(item interface{}, pipe chan<- interface{}) {
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
atomic.AddInt32(&count, int32(item.(int)))
|
||||
}).Done()
|
||||
assert.Equal(t, int32(6), count)
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var count int32
|
||||
Just(1, 2, 3).Walk(func(item interface{}, pipe chan<- interface{}) {
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
atomic.AddInt32(&count, int32(item.(int)))
|
||||
}).Done()
|
||||
assert.Equal(t, int32(6), count)
|
||||
})
|
||||
}
|
||||
|
||||
func TestJust(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
assert.Equal(t, 10, result)
|
||||
})
|
||||
assert.Equal(t, 10, result)
|
||||
}
|
||||
|
||||
func TestDistinct(t *testing.T) {
|
||||
var result int
|
||||
Just(4, 1, 3, 2, 3, 4).Distinct(func(item interface{}) interface{} {
|
||||
return item
|
||||
}).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var result int
|
||||
Just(4, 1, 3, 2, 3, 4).Distinct(func(item interface{}) interface{} {
|
||||
return item
|
||||
}).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
assert.Equal(t, 10, result)
|
||||
})
|
||||
assert.Equal(t, 10, result)
|
||||
}
|
||||
|
||||
func TestFilter(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Filter(func(item interface{}) bool {
|
||||
return item.(int)%2 == 0
|
||||
}).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Filter(func(item interface{}) bool {
|
||||
return item.(int)%2 == 0
|
||||
}).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
assert.Equal(t, 6, result)
|
||||
})
|
||||
}
|
||||
|
||||
func TestFirst(t *testing.T) {
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
assert.Nil(t, Just().First())
|
||||
assert.Equal(t, "foo", Just("foo").First())
|
||||
assert.Equal(t, "foo", Just("foo", "bar").First())
|
||||
})
|
||||
assert.Equal(t, 6, result)
|
||||
}
|
||||
|
||||
func TestForAll(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Filter(func(item interface{}) bool {
|
||||
return item.(int)%2 == 0
|
||||
}).ForAll(func(pipe <-chan interface{}) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Filter(func(item interface{}) bool {
|
||||
return item.(int)%2 == 0
|
||||
}).ForAll(func(pipe <-chan interface{}) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
})
|
||||
assert.Equal(t, 6, result)
|
||||
})
|
||||
assert.Equal(t, 6, result)
|
||||
}
|
||||
|
||||
func TestGroup(t *testing.T) {
|
||||
var groups [][]int
|
||||
Just(10, 11, 20, 21).Group(func(item interface{}) interface{} {
|
||||
v := item.(int)
|
||||
return v / 10
|
||||
}).ForEach(func(item interface{}) {
|
||||
v := item.([]interface{})
|
||||
var group []int
|
||||
for _, each := range v {
|
||||
group = append(group, each.(int))
|
||||
}
|
||||
groups = append(groups, group)
|
||||
})
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var groups [][]int
|
||||
Just(10, 11, 20, 21).Group(func(item interface{}) interface{} {
|
||||
v := item.(int)
|
||||
return v / 10
|
||||
}).ForEach(func(item interface{}) {
|
||||
v := item.([]interface{})
|
||||
var group []int
|
||||
for _, each := range v {
|
||||
group = append(group, each.(int))
|
||||
}
|
||||
groups = append(groups, group)
|
||||
})
|
||||
|
||||
assert.Equal(t, 2, len(groups))
|
||||
for _, group := range groups {
|
||||
assert.Equal(t, 2, len(group))
|
||||
assert.True(t, group[0]/10 == group[1]/10)
|
||||
}
|
||||
assert.Equal(t, 2, len(groups))
|
||||
for _, group := range groups {
|
||||
assert.Equal(t, 2, len(group))
|
||||
assert.True(t, group[0]/10 == group[1]/10)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestHead(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Head(2).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Head(2).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
assert.Equal(t, 3, result)
|
||||
})
|
||||
assert.Equal(t, 3, result)
|
||||
}
|
||||
|
||||
func TestHeadZero(t *testing.T) {
|
||||
assert.Panics(t, func() {
|
||||
Just(1, 2, 3, 4).Head(0).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
return nil, nil
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
assert.Panics(t, func() {
|
||||
Just(1, 2, 3, 4).Head(0).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
return nil, nil
|
||||
})
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func TestHeadMore(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Head(6).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Head(6).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
assert.Equal(t, 10, result)
|
||||
})
|
||||
}
|
||||
|
||||
func TestLast(t *testing.T) {
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
goroutines := runtime.NumGoroutine()
|
||||
assert.Nil(t, Just().Last())
|
||||
assert.Equal(t, "foo", Just("foo").Last())
|
||||
assert.Equal(t, "bar", Just("foo", "bar").Last())
|
||||
// let scheduler schedule first
|
||||
runtime.Gosched()
|
||||
assert.Equal(t, goroutines, runtime.NumGoroutine())
|
||||
})
|
||||
assert.Equal(t, 10, result)
|
||||
}
|
||||
|
||||
func TestMap(t *testing.T) {
|
||||
log.SetOutput(ioutil.Discard)
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
log.SetOutput(ioutil.Discard)
|
||||
|
||||
tests := []struct {
|
||||
mapper MapFunc
|
||||
expect int
|
||||
}{
|
||||
{
|
||||
mapper: func(item interface{}) interface{} {
|
||||
v := item.(int)
|
||||
return v * v
|
||||
tests := []struct {
|
||||
mapper MapFunc
|
||||
expect int
|
||||
}{
|
||||
{
|
||||
mapper: func(item interface{}) interface{} {
|
||||
v := item.(int)
|
||||
return v * v
|
||||
},
|
||||
expect: 30,
|
||||
},
|
||||
expect: 30,
|
||||
},
|
||||
{
|
||||
mapper: func(item interface{}) interface{} {
|
||||
v := item.(int)
|
||||
if v%2 == 0 {
|
||||
return 0
|
||||
}
|
||||
return v * v
|
||||
},
|
||||
expect: 10,
|
||||
},
|
||||
{
|
||||
mapper: func(item interface{}) interface{} {
|
||||
v := item.(int)
|
||||
if v%2 == 0 {
|
||||
panic(v)
|
||||
}
|
||||
return v * v
|
||||
},
|
||||
expect: 10,
|
||||
},
|
||||
}
|
||||
|
||||
// Map(...) works even WithWorkers(0)
|
||||
for i, test := range tests {
|
||||
t.Run(stringx.Rand(), func(t *testing.T) {
|
||||
var result int
|
||||
var workers int
|
||||
if i%2 == 0 {
|
||||
workers = 0
|
||||
} else {
|
||||
workers = runtime.NumCPU()
|
||||
}
|
||||
From(func(source chan<- interface{}) {
|
||||
for i := 1; i < 5; i++ {
|
||||
source <- i
|
||||
}
|
||||
}).Map(test.mapper, WithWorkers(workers)).Reduce(
|
||||
func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
{
|
||||
mapper: func(item interface{}) interface{} {
|
||||
v := item.(int)
|
||||
if v%2 == 0 {
|
||||
return 0
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
return v * v
|
||||
},
|
||||
expect: 10,
|
||||
},
|
||||
{
|
||||
mapper: func(item interface{}) interface{} {
|
||||
v := item.(int)
|
||||
if v%2 == 0 {
|
||||
panic(v)
|
||||
}
|
||||
return v * v
|
||||
},
|
||||
expect: 10,
|
||||
},
|
||||
}
|
||||
|
||||
assert.Equal(t, test.expect, result)
|
||||
})
|
||||
}
|
||||
// Map(...) works even WithWorkers(0)
|
||||
for i, test := range tests {
|
||||
t.Run(stringx.Rand(), func(t *testing.T) {
|
||||
var result int
|
||||
var workers int
|
||||
if i%2 == 0 {
|
||||
workers = 0
|
||||
} else {
|
||||
workers = runtime.NumCPU()
|
||||
}
|
||||
From(func(source chan<- interface{}) {
|
||||
for i := 1; i < 5; i++ {
|
||||
source <- i
|
||||
}
|
||||
}).Map(test.mapper, WithWorkers(workers)).Reduce(
|
||||
func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
|
||||
assert.Equal(t, test.expect, result)
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestMerge(t *testing.T) {
|
||||
Just(1, 2, 3, 4).Merge().ForEach(func(item interface{}) {
|
||||
assert.ElementsMatch(t, []interface{}{1, 2, 3, 4}, item.([]interface{}))
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
Just(1, 2, 3, 4).Merge().ForEach(func(item interface{}) {
|
||||
assert.ElementsMatch(t, []interface{}{1, 2, 3, 4}, item.([]interface{}))
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func TestParallelJust(t *testing.T) {
|
||||
var count int32
|
||||
Just(1, 2, 3).Parallel(func(item interface{}) {
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
atomic.AddInt32(&count, int32(item.(int)))
|
||||
}, UnlimitedWorkers())
|
||||
assert.Equal(t, int32(6), count)
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var count int32
|
||||
Just(1, 2, 3).Parallel(func(item interface{}) {
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
atomic.AddInt32(&count, int32(item.(int)))
|
||||
}, UnlimitedWorkers())
|
||||
assert.Equal(t, int32(6), count)
|
||||
})
|
||||
}
|
||||
|
||||
func TestReverse(t *testing.T) {
|
||||
Just(1, 2, 3, 4).Reverse().Merge().ForEach(func(item interface{}) {
|
||||
assert.ElementsMatch(t, []interface{}{4, 3, 2, 1}, item.([]interface{}))
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
Just(1, 2, 3, 4).Reverse().Merge().ForEach(func(item interface{}) {
|
||||
assert.ElementsMatch(t, []interface{}{4, 3, 2, 1}, item.([]interface{}))
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func TestSort(t *testing.T) {
|
||||
var prev int
|
||||
Just(5, 3, 7, 1, 9, 6, 4, 8, 2).Sort(func(a, b interface{}) bool {
|
||||
return a.(int) < b.(int)
|
||||
}).ForEach(func(item interface{}) {
|
||||
next := item.(int)
|
||||
assert.True(t, prev < next)
|
||||
prev = next
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var prev int
|
||||
Just(5, 3, 7, 1, 9, 6, 4, 8, 2).Sort(func(a, b interface{}) bool {
|
||||
return a.(int) < b.(int)
|
||||
}).ForEach(func(item interface{}) {
|
||||
next := item.(int)
|
||||
assert.True(t, prev < next)
|
||||
prev = next
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func TestSplit(t *testing.T) {
|
||||
assert.Panics(t, func() {
|
||||
Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).Split(0).Done()
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
assert.Panics(t, func() {
|
||||
Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).Split(0).Done()
|
||||
})
|
||||
var chunks [][]interface{}
|
||||
Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).Split(4).ForEach(func(item interface{}) {
|
||||
chunk := item.([]interface{})
|
||||
chunks = append(chunks, chunk)
|
||||
})
|
||||
assert.EqualValues(t, [][]interface{}{
|
||||
{1, 2, 3, 4},
|
||||
{5, 6, 7, 8},
|
||||
{9, 10},
|
||||
}, chunks)
|
||||
})
|
||||
var chunks [][]interface{}
|
||||
Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).Split(4).ForEach(func(item interface{}) {
|
||||
chunk := item.([]interface{})
|
||||
chunks = append(chunks, chunk)
|
||||
})
|
||||
assert.EqualValues(t, [][]interface{}{
|
||||
{1, 2, 3, 4},
|
||||
{5, 6, 7, 8},
|
||||
{9, 10},
|
||||
}, chunks)
|
||||
}
|
||||
|
||||
func TestTail(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Tail(2).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4).Tail(2).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
for item := range pipe {
|
||||
result += item.(int)
|
||||
}
|
||||
return result, nil
|
||||
})
|
||||
assert.Equal(t, 7, result)
|
||||
})
|
||||
assert.Equal(t, 7, result)
|
||||
}
|
||||
|
||||
func TestTailZero(t *testing.T) {
|
||||
assert.Panics(t, func() {
|
||||
Just(1, 2, 3, 4).Tail(0).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
return nil, nil
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
assert.Panics(t, func() {
|
||||
Just(1, 2, 3, 4).Tail(0).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
|
||||
return nil, nil
|
||||
})
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func TestWalk(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4, 5).Walk(func(item interface{}, pipe chan<- interface{}) {
|
||||
if item.(int)%2 != 0 {
|
||||
pipe <- item
|
||||
}
|
||||
}, UnlimitedWorkers()).ForEach(func(item interface{}) {
|
||||
result += item.(int)
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
var result int
|
||||
Just(1, 2, 3, 4, 5).Walk(func(item interface{}, pipe chan<- interface{}) {
|
||||
if item.(int)%2 != 0 {
|
||||
pipe <- item
|
||||
}
|
||||
}, UnlimitedWorkers()).ForEach(func(item interface{}) {
|
||||
result += item.(int)
|
||||
})
|
||||
assert.Equal(t, 9, result)
|
||||
})
|
||||
}
|
||||
|
||||
func TestStream_AnyMach(t *testing.T) {
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
|
||||
return item.(int) == 4
|
||||
}))
|
||||
assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
|
||||
return item.(int) == 0
|
||||
}))
|
||||
assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
|
||||
return item.(int) == 2
|
||||
}))
|
||||
assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
|
||||
return item.(int) == 2
|
||||
}))
|
||||
})
|
||||
}
|
||||
|
||||
func TestStream_AllMach(t *testing.T) {
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
assetEqual(
|
||||
t, true, Just(1, 2, 3).AllMach(func(item interface{}) bool {
|
||||
return true
|
||||
}),
|
||||
)
|
||||
assetEqual(
|
||||
t, false, Just(1, 2, 3).AllMach(func(item interface{}) bool {
|
||||
return false
|
||||
}),
|
||||
)
|
||||
assetEqual(
|
||||
t, false, Just(1, 2, 3).AllMach(func(item interface{}) bool {
|
||||
return item.(int) == 1
|
||||
}),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
func TestStream_NoneMatch(t *testing.T) {
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
assetEqual(
|
||||
t, true, Just(1, 2, 3).NoneMatch(func(item interface{}) bool {
|
||||
return false
|
||||
}),
|
||||
)
|
||||
assetEqual(
|
||||
t, false, Just(1, 2, 3).NoneMatch(func(item interface{}) bool {
|
||||
return true
|
||||
}),
|
||||
)
|
||||
assetEqual(
|
||||
t, true, Just(1, 2, 3).NoneMatch(func(item interface{}) bool {
|
||||
return item.(int) == 4
|
||||
}),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
func TestConcat(t *testing.T) {
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
a1 := []interface{}{1, 2, 3}
|
||||
a2 := []interface{}{4, 5, 6}
|
||||
s1 := Just(a1...)
|
||||
s2 := Just(a2...)
|
||||
stream := Concat(s1, s2)
|
||||
var items []interface{}
|
||||
for item := range stream.source {
|
||||
items = append(items, item)
|
||||
}
|
||||
sort.Slice(items, func(i, j int) bool {
|
||||
return items[i].(int) < items[j].(int)
|
||||
})
|
||||
ints := make([]interface{}, 0)
|
||||
ints = append(ints, a1...)
|
||||
ints = append(ints, a2...)
|
||||
assetEqual(t, ints, items)
|
||||
})
|
||||
}
|
||||
|
||||
func TestStream_Skip(t *testing.T) {
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
assetEqual(t, 3, Just(1, 2, 3, 4).Skip(1).Count())
|
||||
assetEqual(t, 1, Just(1, 2, 3, 4).Skip(3).Count())
|
||||
assetEqual(t, 4, Just(1, 2, 3, 4).Skip(0).Count())
|
||||
equal(t, Just(1, 2, 3, 4).Skip(3), []interface{}{4})
|
||||
assert.Panics(t, func() {
|
||||
Just(1, 2, 3, 4).Skip(-1)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func TestStream_Concat(t *testing.T) {
|
||||
runCheckedTest(t, func(t *testing.T) {
|
||||
stream := Just(1).Concat(Just(2), Just(3))
|
||||
var items []interface{}
|
||||
for item := range stream.source {
|
||||
items = append(items, item)
|
||||
}
|
||||
sort.Slice(items, func(i, j int) bool {
|
||||
return items[i].(int) < items[j].(int)
|
||||
})
|
||||
assetEqual(t, []interface{}{1, 2, 3}, items)
|
||||
|
||||
just := Just(1)
|
||||
equal(t, just.Concat(just), []interface{}{1})
|
||||
})
|
||||
assert.Equal(t, 9, result)
|
||||
}
|
||||
|
||||
func BenchmarkParallelMapReduce(b *testing.B) {
|
||||
@@ -377,6 +546,12 @@ func BenchmarkMapReduce(b *testing.B) {
|
||||
}).Map(mapper).Reduce(reducer)
|
||||
}
|
||||
|
||||
func assetEqual(t *testing.T, except, data interface{}) {
|
||||
if !reflect.DeepEqual(except, data) {
|
||||
t.Errorf(" %v, want %v", data, except)
|
||||
}
|
||||
}
|
||||
|
||||
func equal(t *testing.T, stream Stream, data []interface{}) {
|
||||
items := make([]interface{}, 0)
|
||||
for item := range stream.source {
|
||||
@@ -387,85 +562,10 @@ func equal(t *testing.T, stream Stream, data []interface{}) {
|
||||
}
|
||||
}
|
||||
|
||||
func assetEqual(t *testing.T, except, data interface{}) {
|
||||
if !reflect.DeepEqual(except, data) {
|
||||
t.Errorf(" %v, want %v", data, except)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStream_AnyMach(t *testing.T) {
|
||||
assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
|
||||
return item.(int) == 4
|
||||
}))
|
||||
assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
|
||||
return item.(int) == 0
|
||||
}))
|
||||
assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
|
||||
return item.(int) == 2
|
||||
}))
|
||||
assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item interface{}) bool {
|
||||
return item.(int) == 2
|
||||
}))
|
||||
}
|
||||
|
||||
func TestStream_AllMach(t *testing.T) {
|
||||
assetEqual(
|
||||
t, true, Just(1, 2, 3).AllMach(func(item interface{}) bool {
|
||||
return true
|
||||
}),
|
||||
)
|
||||
assetEqual(
|
||||
t, false, Just(1, 2, 3).AllMach(func(item interface{}) bool {
|
||||
return false
|
||||
}),
|
||||
)
|
||||
assetEqual(
|
||||
t, false, Just(1, 2, 3).AllMach(func(item interface{}) bool {
|
||||
return item.(int) == 1
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
func TestConcat(t *testing.T) {
|
||||
a1 := []interface{}{1, 2, 3}
|
||||
a2 := []interface{}{4, 5, 6}
|
||||
s1 := Just(a1...)
|
||||
s2 := Just(a2...)
|
||||
stream := Concat(s1, s2)
|
||||
var items []interface{}
|
||||
for item := range stream.source {
|
||||
items = append(items, item)
|
||||
}
|
||||
sort.Slice(items, func(i, j int) bool {
|
||||
return items[i].(int) < items[j].(int)
|
||||
})
|
||||
ints := make([]interface{}, 0)
|
||||
ints = append(ints, a1...)
|
||||
ints = append(ints, a2...)
|
||||
assetEqual(t, ints, items)
|
||||
}
|
||||
|
||||
func TestStream_Skip(t *testing.T) {
|
||||
assetEqual(t, 3, Just(1, 2, 3, 4).Skip(1).Count())
|
||||
assetEqual(t, 1, Just(1, 2, 3, 4).Skip(3).Count())
|
||||
assetEqual(t, 4, Just(1, 2, 3, 4).Skip(0).Count())
|
||||
equal(t, Just(1, 2, 3, 4).Skip(3), []interface{}{4})
|
||||
assert.Panics(t, func() {
|
||||
Just(1, 2, 3, 4).Skip(-1)
|
||||
})
|
||||
}
|
||||
|
||||
func TestStream_Concat(t *testing.T) {
|
||||
stream := Just(1).Concat(Just(2), Just(3))
|
||||
var items []interface{}
|
||||
for item := range stream.source {
|
||||
items = append(items, item)
|
||||
}
|
||||
sort.Slice(items, func(i, j int) bool {
|
||||
return items[i].(int) < items[j].(int)
|
||||
})
|
||||
assetEqual(t, []interface{}{1, 2, 3}, items)
|
||||
|
||||
just := Just(1)
|
||||
equal(t, just.Concat(just), []interface{}{1})
|
||||
func runCheckedTest(t *testing.T, fn func(t *testing.T)) {
|
||||
goroutines := runtime.NumGoroutine()
|
||||
fn(t)
|
||||
// let scheduler schedule first
|
||||
time.Sleep(time.Millisecond)
|
||||
assert.Equal(t, goroutines, runtime.NumGoroutine())
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user