func (s *StreamWrapper) Close() error {
	return s.stream.Close()
}
Close stream and shut down all process
package eventstream
import (
	"context"
	"strings"
	"github.com/Knetic/govaluate"
	"github.com/demdxx/gocast"
)
type StreamWrapper struct {
	stream Streamer
	whereCondition *govaluate.EvaluableExpression
}
func NewStreamWrapper(stream Streamer, where string) (_ Streamer, err error) {
	var whereObj *govaluate.EvaluableExpression
	if len(strings.TrimSpace(where)) > 0 {
		if whereObj, err = govaluate.NewEvaluableExpression(where); err != nil {
			return
		}
	}
	if whereObj == nil {
		return stream, nil
	}
	return &StreamWrapper{
		stream:         stream,
		whereCondition: whereObj,
	}, nil
}
func (s *StreamWrapper) ID() string {
	return s.stream.ID()
}
func (s *StreamWrapper) Put(ctx context.Context, msg Message) error {
	return s.stream.Put(ctx, msg)
}
func (s *StreamWrapper) Check(ctx context.Context, msg Message) bool {
	if !s.stream.Check(ctx, msg) {
		return false
	}
	if s.whereCondition != nil {
		r, err := s.whereCondition.Evaluate(msg.Map())
		return err == nil && gocast.ToBool(r)
	}
	return true
}
func (s *StreamWrapper) Run(ctx context.Context) error {
	return s.stream.Run(ctx)
}