Function TransformSimple

pkg/stream/stream.go:572–603

TransformSimple 创建新的 goroutine 进行转换

(r *Reader[T], fn func(T) (U, error))

// TransformSimple 创建新的 goroutine 进行转换
func TransformSimple[T, U any](r *Reader[T], fn func(T) (U, error)) *Reader[U] {
	out, writer := Pipe[U](5)

	go func() {
		defer writer.Close()
		for {
			v, err := r.Recv()
			if errors.Is(err, io.EOF) {
				return
			}
			if err != nil {
				writer.Send(*new(U), err)
				return
			}

			result, err := fn(v)
			if errors.Is(err, ErrSkip) {
				continue
			}
			if err != nil {
				writer.Send(*new(U), err)
				return
			}

			if writer.Send(result, nil) {
				return
			}
		}
	}()

	return out
}

// Filter 创建只包含匹配谓词的值的新读取器
func Filter[T any](r *Reader[T], predicate func(T) bool) *Reader[T] {