func FilterFunc(f interface{}) (operators.UnFunc, error) {
	fntype := reflect.TypeOf(f)

	funcForm, err := isUnaryFuncForm(fntype)
	if err != nil {
		return nil, err
	}
	if funcForm == unaryFuncUnsupported {
		return nil, fmt.Errorf("unsupported unary func type")
	}

	// ensure bool ret type
	if fntype.Out(0).Kind() != reflect.Bool {
		return nil, fmt.Errorf("unary filter func must return bool")
	}

	fnval := reflect.ValueOf(f)
	return operators.UnFunc(func(ctx api.StreamContext, data interface{}) interface{} {
		result := callOpFunc(fnval, ctx, data, funcForm)
		predicate := result.Bool()
		if !predicate {
			return nil
		}
		return data
	}), nil
}
// FilterFunc returns a unary function (api.UnFunc) which applies the user-defined
// filtering to apply predicates that filters out data items from being included
// in the downstream.  The provided user-defined function must be of type:
//   func(T)bool - where T is the type of incoming data item, bool is the value of the predicate
// When the user-defined function returns false, the current processed data item will not
// be placed in the downstream processing.
FilterFunc returns a unary function (api.UnFunc) which applies the user-defined
filtering to apply predicates that filters out data items from being included
in the downstream.  The provided user-defined function must be of type:
func(T)bool - where T is the type of incoming data item, bool is the value of the predicate
When the user-defined function returns false, the current processed data item will not
be placed in the downstream processing.
// FilterFunc returns a unary function ( api . UnFunc ) which applies the user - defined // filtering to apply predicates that filters out data items from being included // in the downstream . The provided user - defined function must be of type : // func ( T ) bool - where T is the type of incoming data item bool is the value of the predicate // When the user - defined function returns false the current processed data item will not // be placed in the downstream processing .
func FilterFunc(f interface{}) (operators.UnFunc, error) {
	fntype := reflect.TypeOf(f)
	funcForm, err := isUnaryFuncForm(fntype)
	if err != nil {
		return nil, err
	}
	if funcForm == unaryFuncUnsupported {
		return nil, fmt.Errorf("unsupported unary func type")
	}
	if fntype.Out(0).Kind() != reflect.Bool {
		return nil, fmt.Errorf("unary filter func must return bool")
	}
	fnval := reflect.ValueOf(f)
	return operators.UnFunc(func(ctx api.StreamContext, data interface{}) interface{} {
		result := callOpFunc(fnval, ctx, data, funcForm)
		predicate := result.Bool()
		if !predicate {
			return nil
		}
		return data
	}), nil
}
func FilterFunc ( f interface { } ) ( operators . UnFunc , error ) { fntype := reflect . TypeOf ( f ) funcForm , err := isUnaryFuncForm ( fntype ) if err != nil { return nil , err } if funcForm == unaryFuncUnsupported { return nil , fmt . Errorf ( " " ) } if fntype . Out ( 0 ) . Kind ( ) != reflect . Bool { return nil , fmt . Errorf ( " " ) } fnval := reflect . ValueOf ( f ) return operators . UnFunc ( func ( ctx api . StreamContext , data interface { } ) interface { } { result := callOpFunc ( fnval , ctx , data , funcForm ) predicate := result . Bool ( ) if ! predicate { return nil } return data } ) , nil }
FilterFunc returns a unary function (api.UnFunc) which applies the user-defined
filtering to apply predicates that filters out data items from being included
in the downstream.
FilterFunc returns a unary function ( api . UnFunc ) which applies the user - defined filtering to apply predicates that filters out data items from being included in the downstream .
// ensure bool ret type