Function StreamFilter

pkg/agent/streaming.go:233–256

StreamFilter 辅助函数 - 过滤事件

(reader *stream.Reader[*session.Event], predicate func(*session.Event) bool)

// StreamFilter 辅助函数 - 过滤事件
func StreamFilter(reader *stream.Reader[*session.Event], predicate func(*session.Event) bool) *stream.Reader[*session.Event] {
	outReader, outWriter := stream.Pipe[*session.Event](10)

	go func() {
		defer outWriter.Close()
		for {
			event, err := reader.Recv()
			if err != nil {
				if errors.Is(err, io.EOF) {
					break
				}
				outWriter.Send(nil, err)
				return
			}
			if event != nil && predicate(event) {
				if outWriter.Send(event, nil) {
					return
				}
			}
		}
	}()

	return outReader
}

// streamConfig 流式执行配置
type streamConfig struct {