wp-go/stream/simpleStream.go

package stream

import (
	"github.com/fthvgb1/wp-go/helper"
	"github.com/fthvgb1/wp-go/safety"
	"github.com/fthvgb1/wp-go/taskPools"
)

func SimpleParallelFilterAndMap[R, T any](a SimpleSliceStream[T], fn func(T) (R, bool), c int) SimpleSliceStream[R] {
	var x []R
	rr := safety.NewSlice(x)
	a.ParallelForEach(func(t T) {
		y, ok := fn(t)
		if ok {
			rr.Append(y)
		}
	}, c)
	return SimpleSliceStream[R]{rr.Load()}
}

func SimpleParallelFilterAndMapToMap[K comparable, V any, T any](a SimpleSliceStream[T], fn func(t T) (K, V, bool), c int) (r SimpleMapStream[K, V]) {
	m := newMapX[K, V]()
	a.ParallelForEach(func(t T) {
		k, v, ok := fn(t)
		if ok {
			m.set(k, v)
		}
	}, c)
	var mm = map[K]V{}
	r = NewSimpleMapStream(mm)
	return
}

func SimpleSliceFilterAndMapToMap[K comparable, V any, T any](a SimpleSliceStream[T], fn func(t T) (K, V, bool), isCoverPrev bool) (r SimpleMapStream[K, V]) {
	m := make(map[K]V)
	a.ForEach(func(t T) {
		k, v, ok := fn(t)
		if ok {
			_, ok = m[k]
			if isCoverPrev || !ok {
				m[k] = v
			}
		}
	})
	r.m = m
	return
}

func SimpleStreamFilterAndMap[R, T any](a SimpleSliceStream[T], fn func(T) (R, bool)) SimpleSliceStream[R] {
	return NewSimpleSliceStream(helper.SliceFilterAndMap(a.arr, fn))
}

func SimpleParallelMap[R, T any](a SimpleSliceStream[T], fn func(T) R, c int) SimpleSliceStream[R] {
	var x []R
	rr := safety.NewSlice(x)
	a.ParallelForEach(func(t T) {
		rr.Append(fn(t))
	}, c)
	return SimpleSliceStream[R]{rr.Load()}
}
func SimpleStreamMap[R, T any](a SimpleSliceStream[T], fn func(T) R) SimpleSliceStream[R] {
	return NewSimpleSliceStream(helper.SliceMap(a.arr, fn))
}

func Reduce[T any, S any](s SimpleSliceStream[S], fn func(S, T) T, init T) (r T) {
	return helper.SliceReduce(s.arr, fn, init)
}

func NewSimpleSliceStream[T any](arr []T) SimpleSliceStream[T] {
	return SimpleSliceStream[T]{arr: arr}
}

type SimpleSliceStream[T any] struct {
	arr []T
}

func (r SimpleSliceStream[T]) ForEach(fn func(T)) {
	for _, t := range r.arr {
		fn(t)
	}
}

func (r SimpleSliceStream[T]) ParallelForEach(fn func(T), c int) {
	p := taskPools.NewPools(c)
	for _, t := range r.arr {
		t := t
		p.Execute(func() {
			fn(t)
		})
	}
	p.Wait()
}

func (r SimpleSliceStream[T]) ParallelFilter(fn func(T) bool, c int) SimpleSliceStream[T] {
	rr := safety.NewSlice([]T{})
	r.ParallelForEach(func(t T) {
		if fn(t) {
			rr.Append(t)
		}
	}, c)
	return SimpleSliceStream[T]{rr.Load()}
}
func (r SimpleSliceStream[T]) Filter(fn func(T) bool) SimpleSliceStream[T] {
	r.arr = helper.SliceFilter(r.arr, fn)
	return r
}

func (r SimpleSliceStream[T]) ParallelMap(fn func(T) T, c int) SimpleSliceStream[T] {
	rr := safety.NewSlice([]T{})
	r.ParallelForEach(func(t T) {
		rr.Append(fn(t))
	}, c)
	return SimpleSliceStream[T]{rr.Load()}
}

func (r SimpleSliceStream[T]) Map(fn func(T) T) SimpleSliceStream[T] {
	r.arr = helper.SliceMap(r.arr, fn)
	return r
}

func (r SimpleSliceStream[T]) Sort(fn func(i, j T) bool) SimpleSliceStream[T] {
	helper.SimpleSort(r.arr, fn)
	return r
}

func (r SimpleSliceStream[T]) Len() int {
	return len(r.arr)
}

func (r SimpleSliceStream[T]) Limit(limit, offset int) SimpleSliceStream[T] {
	l := len(r.arr)
	if offset >= l {
		return SimpleSliceStream[T]{}
	}
	ll := offset + limit
	if ll > l {
		ll = l
	}
	return SimpleSliceStream[T]{r.arr[offset:ll]}
}

func (r SimpleSliceStream[T]) Reverse() SimpleSliceStream[T] {
	helper.SliceSelfReverse(r.arr)
	return r
}

func (r SimpleSliceStream[T]) Result() []T {
	return r.arr
}