添加一个简单的 Stream 处理器，可恶不支持泛型方法

helper/map.go

@@ -2,7 +2,7 @@ package helper
 
 import "encoding/json"
 
-func MapToStruct[T any](m map[string]any) (r T, err error) {
+func MapToStruct[T any, M any](m M) (r T, err error) {
 	str, err := json.Marshal(m)
 	if err != nil {
 		return
@@ -30,3 +30,22 @@ func MapToSlice[T any, K comparable, V any](m map[K]V, fn func(K, V) (T, bool))
 	}
 	return
 }
+
+func MapAnyToString(m map[any]any) (r map[string]any) {
+	r = make(map[string]any)
+	for k, v := range m {
+		kk, ok := k.(string)
+		if ok {
+			vv, ok := v.(map[any]any)
+			if ok {
+				x := make(map[string]any)
+				MapAnyToString(vv)
+				r[kk] = x
+			} else {
+				r[kk] = v
+			}
+		}
+
+	}
+	return
+}

helper/slice.go

@@ -8,6 +8,16 @@ func SliceMap[T, R any](arr []T, fn func(T) R) []R {
 	return r
 }
 
+func SliceFilterAndMap[N any, T any](arr []T, fn func(T) (N, bool)) (r []N) {
+	for _, t := range arr {
+		x, ok := fn(t)
+		if ok {
+			r = append(r, x)
+		}
+	}
+	return
+}
+
 func SliceFilter[T any](arr []T, fn func(T) bool) []T {
 	var r []T
 	for _, t := range arr {
@@ -18,7 +28,7 @@ func SliceFilter[T any](arr []T, fn func(T) bool) []T {
 	return r
 }
 
-func SliceReduce[T, R any](arr []T, fn func(T, R) R, r R) R {
+func SliceReduce[R, T any](arr []T, fn func(T, R) R, r R) R {
 	for _, t := range arr {
 		r = fn(t, r)
 	}

helper/slice_test.go

@@ -477,3 +477,45 @@ func TestSliceSelfReverse(t *testing.T) {
 		})
 	}
 }
+
+func TestSliceFilterAndMap(t *testing.T) {
+	type a struct {
+		x int
+		y string
+	}
+
+	type args[T any, N any] struct {
+		arr []T
+		fn  func(T) (N, bool)
+	}
+	type testCase[T any, N any] struct {
+		name  string
+		args  args[T, N]
+		wantR []N
+	}
+	tests := []testCase[a, string]{
+		{
+			name: "t1",
+			args: args[a, string]{
+				arr: []a{
+					{1, "1"}, {2, "2"}, {3, "3"},
+				},
+				fn: func(t a) (r string, ok bool) {
+					if t.x > 2 {
+						r = t.y
+						ok = true
+					}
+					return
+				},
+			},
+			wantR: []string{"3"},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if gotR := SliceFilterAndMap[string](tt.args.arr, tt.args.fn); !reflect.DeepEqual(gotR, tt.wantR) {
+				t.Errorf("SliceFilterAndMap() = %v, want %v", gotR, tt.wantR)
+			}
+		})
+	}
+}

safety/slice.go

@@ -0,0 +1,22 @@
+package safety
+
+import "sync"
+
+type Slice[T any] struct {
+	Var[[]T]
+	mu sync.Mutex
+}
+
+func NewSlice[T any](a []T) *Slice[T] {
+	return &Slice[T]{
+		NewVar(a),
+		sync.Mutex{},
+	}
+}
+
+func (r *Slice[T]) Append(t ...T) {
+	r.mu.Lock()
+	ts := append(r.Load(), t...)
+	r.Store(ts)
+	r.mu.Unlock()
+}

safety/slice_test.go

@@ -0,0 +1,39 @@
+package safety
+
+import (
+	"fmt"
+	"github/fthvgb1/wp-go/helper"
+	"testing"
+	"time"
+)
+
+func TestSlice_Append(t *testing.T) {
+	type args[T any] struct {
+		t []T
+	}
+	type testCase[T any] struct {
+		name string
+		r    Slice[T]
+		args args[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    *NewSlice([]int{}),
+			args: args[int]{helper.RangeSlice(1, 10, 1)},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fn := func() {
+				tt.r.Append(tt.args.t...)
+			}
+			go fn()
+			go fn()
+			go fn()
+			go fn()
+			time.Sleep(time.Second)
+			fmt.Println(tt.r.Load())
+		})
+	}
+}

stream/simpleStream.go

@@ -0,0 +1,92 @@
+package stream
+
+import (
+	"github/fthvgb1/wp-go/helper"
+	"github/fthvgb1/wp-go/safety"
+	"github/fthvgb1/wp-go/taskPools"
+)
+
+func Reduce[T any, S any](s SimpleSliceStream[S], fn func(S, T) T) (r T) {
+	return helper.SliceReduce(s.arr, fn, r)
+}
+
+type SimpleSliceStream[T any] struct {
+	arr []T
+}
+
+func NewSimpleSliceStream[T any](arr []T) SimpleSliceStream[T] {
+	return SimpleSliceStream[T]{arr: arr}
+}
+
+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] {
+	p := taskPools.NewPools(c)
+	var x []T
+	rr := safety.NewSlice(x)
+	for _, t := range r.arr {
+		t := t
+		p.Execute(func() {
+			if fn(t) {
+				rr.Append(t)
+			}
+		})
+	}
+	p.Wait()
+	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] {
+	p := taskPools.NewPools(c)
+	var x []T
+	rr := safety.NewSlice(x)
+	for _, t := range r.arr {
+		t := t
+		p.Execute(func() {
+			rr.Append(fn(t))
+		})
+	}
+	p.Wait()
+	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]) Limit(limit, offset int) SimpleSliceStream[T] {
+	return SimpleSliceStream[T]{r.arr[offset : offset+limit]}
+}
+
+func (r SimpleSliceStream[T]) Reverse() SimpleSliceStream[T] {
+	helper.SliceSelfReverse(r.arr)
+	return r
+}
+
+func (r SimpleSliceStream[T]) Result() []T {
+	return r.arr
+}

stream/simpleStream_test.go

@@ -0,0 +1,337 @@
+package stream
+
+import (
+	"fmt"
+	"github/fthvgb1/wp-go/helper"
+	"reflect"
+	"testing"
+)
+
+var s = NewSimpleSliceStream(helper.RangeSlice(1, 10, 1))
+
+func TestSimpleSliceStream_Filter(t *testing.T) {
+	type args[T int] struct {
+		fn func(T) bool
+	}
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		args args[T]
+		want SimpleSliceStream[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			args: args[int]{
+				func(t int) (r bool) {
+					if t > 5 {
+						r = true
+					}
+					return
+				},
+			},
+			want: SimpleSliceStream[int]{helper.RangeSlice(6, 10, 1)},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.r.Filter(tt.args.fn); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Filter() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSimpleSliceStream_ForEach(t *testing.T) {
+	type args[T int] struct {
+		fn func(T)
+	}
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		args args[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			args: args[int]{
+				func(t int) {
+					fmt.Println(t)
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.r.ForEach(tt.args.fn)
+		})
+	}
+}
+
+func TestSimpleSliceStream_Limit(t *testing.T) {
+	type args struct {
+		limit  int
+		offset int
+	}
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		args args
+		want SimpleSliceStream[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			args: args{
+				limit:  3,
+				offset: 5,
+			},
+			want: SimpleSliceStream[int]{helper.RangeSlice(6, 8, 1)},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.r.Limit(tt.args.limit, tt.args.offset); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Limit() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSimpleSliceStream_Map(t *testing.T) {
+	type args[T int] struct {
+		fn func(T) T
+	}
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		args args[T]
+		want SimpleSliceStream[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			args: args[int]{
+				func(t int) (r int) {
+					return t * 2
+				},
+			},
+			want: SimpleSliceStream[int]{helper.RangeSlice(2, 20, 2)},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.r.Map(tt.args.fn); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Map() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSimpleSliceStream_Result(t *testing.T) {
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		want []T
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			want: helper.RangeSlice(1, 10, 1),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.r.Result(); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Result() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSimpleSliceStream_Sort(t *testing.T) {
+	type args[T int] struct {
+		fn func(i, j T) bool
+	}
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		args args[T]
+		want SimpleSliceStream[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			args: args[int]{
+				fn: func(i, j int) bool {
+					return i > j
+				},
+			},
+			want: SimpleSliceStream[int]{helper.RangeSlice(10, 1, -1)},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.r.Sort(tt.args.fn); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Sort() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSimpleSliceStream_parallelForEach(t *testing.T) {
+	type args[T int] struct {
+		fn func(T)
+		c  int
+	}
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		args args[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			args: args[int]{
+				fn: func(t int) {
+					fmt.Println(t)
+				},
+				c: 3,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.r.ParallelForEach(tt.args.fn, tt.args.c)
+		})
+	}
+}
+
+func TestSimpleSliceStream_ParallelFilter(t *testing.T) {
+	type args[T int] struct {
+		fn func(T) bool
+		c  int
+	}
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		args args[T]
+		want SimpleSliceStream[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			args: args[int]{
+				fn: func(t int) bool {
+					return t > 3
+				},
+				c: 6,
+			},
+			want: SimpleSliceStream[int]{helper.RangeSlice(4, 10, 1)},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.r.ParallelFilter(tt.args.fn, tt.args.c).Sort(func(i, j int) bool {
+				return i < j
+			}); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ParallelFilter() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSimpleSliceStream_ParallelMap(t *testing.T) {
+	type args[T int] struct {
+		fn func(T) T
+		c  int
+	}
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		args args[T]
+		want SimpleSliceStream[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			args: args[int]{
+				fn: func(t int) int {
+					return t * 2
+				},
+				c: 6,
+			},
+			want: SimpleSliceStream[int]{helper.RangeSlice(2, 20, 2)},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.r.ParallelMap(tt.args.fn, tt.args.c).Sort(func(i, j int) bool {
+				return i < j
+			}); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ParallelMap() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestReduce(t *testing.T) {
+	type args[S, T int] struct {
+		s  SimpleSliceStream[S]
+		fn func(S, T) T
+	}
+	type testCase[S, T int] struct {
+		name  string
+		args  args[S, T]
+		wantR T
+	}
+	tests := []testCase[int, int]{
+		{
+			name: "t1",
+			args: args[int, int]{
+				s, func(i, r int) int {
+					return i + r
+				},
+			},
+			wantR: helper.Sum(s.Result()...),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if gotR := Reduce(tt.args.s, tt.args.fn); !reflect.DeepEqual(gotR, tt.wantR) {
+				t.Errorf("Reduce() = %v, want %v", gotR, tt.wantR)
+			}
+		})
+	}
+}
+
+func TestSimpleSliceStream_Reverse(t *testing.T) {
+	type testCase[T int] struct {
+		name string
+		r    SimpleSliceStream[T]
+		want SimpleSliceStream[T]
+	}
+	tests := []testCase[int]{
+		{
+			name: "t1",
+			r:    s,
+			want: SimpleSliceStream[int]{helper.RangeSlice(10, 1, -1)},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.r.Reverse(); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Reverse() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}