func (lambda *Lambda) Element() (runtime.Object, error) {
	var element runtime.Object
	for item := range lambda.val {
		element = item
	}
	if element == nil {
		return nil, fmt.Errorf("no element found")
	}
	return element, nil
}
Element returns a single element. Note that if multiple elements
exists in the lambda pipeline, a random object may be returned.
package lambda
import (
	"fmt"
	"reflect"
	"strings"
	"time"
	"k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/dynamic"
)
type Predicate interface{}
type Consumer interface{}
type Function interface{}
type Producer interface{}
func callPredicate(f interface{}, arg interface{}) bool {
	if isZeroOfUnderlyingType(arg) {
		panic(fmt.Sprintf("nil argument detected when calling predicate %#v with arg %#v", f, arg))
	}
	ret := reflect.ValueOf(f).Call([]reflect.Value{
		reflect.ValueOf(arg),
	})
	return ret[0].Bool()
}
func callFunction(f interface{}, arg interface{}) {
	if isZeroOfUnderlyingType(arg) {
		panic(fmt.Sprintf("nil argument detected when calling function %#v with arg %#v", f, arg))
	}
	reflect.ValueOf(f).Call([]reflect.Value{
		reflect.ValueOf(arg),
	})
}
func callConsumer(f interface{}, arg interface{}) interface{} {
	if isZeroOfUnderlyingType(arg) {
		panic(fmt.Sprintf("nil argument detected when calling consumer %#v with arg %#v", f, arg))
	}
	ret := reflect.ValueOf(f).Call([]reflect.Value{
		reflect.ValueOf(arg),
	})
	if isZeroOfUnderlyingType(ret[0].Interface()) {
		panic(fmt.Sprintf("nil return value detected when calling consumer %#v with arg %#v", f, arg))
	}
	return ret[0].Interface()
}
func callProducer(f interface{}) interface{} {
	ret := reflect.ValueOf(f).Call([]reflect.Value{})
	if isZeroOfUnderlyingType(ret[0].Interface()) {
		panic(fmt.Sprintf("nil return value detected when calling producer %#v", f))
	}
	return ret[0].Interface()
}
type Lambda struct {
	getFunc    func(namespace, name string) (runtime.Object, error)
	listFunc   func(namespace string, selector labels.Selector) ([]runtime.Object, error)
	createFunc func(runtime.Object) error
	updateFunc func(runtime.Object) error
	deleteFunc func(runtime.Object) error
	clientInterface dynamic.Interface
	rs              Resource
	namespaces      []string
	val             <-chan runtime.Object
	Errors          []error
}
func (lambda *Lambda) run(f func()) error {
	if !lambda.NoError() {
		return &ErrMultiLambdaFailure{
			errors: lambda.Errors,
		}
	}
	f()
	if len(lambda.Errors) != 0 {
		return &ErrMultiLambdaFailure{
			errors: lambda.Errors,
		}
	}
	return nil
}
func (lambda *Lambda) addError(err error) {
	if lambda.Errors == nil {
		lambda.Errors = []error{err}
	}
	lambda.Errors = append(lambda.Errors, err)
}
func (lambda *Lambda) clone() (*Lambda, chan runtime.Object) {
	ch := make(chan runtime.Object)
	l := &Lambda{
		rs:              lambda.rs,
		val:             ch,
		Errors:          lambda.Errors,
		getFunc:         lambda.getFunc,
		listFunc:        lambda.listFunc,
		createFunc:      lambda.createFunc,
		updateFunc:      lambda.updateFunc,
		deleteFunc:      lambda.deleteFunc,
		clientInterface: lambda.clientInterface,
	}
	return l, ch
}
func (lambda *Lambda) Collect() *Lambda {
	l, ch := lambda.clone()
	go func() {
		defer close(ch)
		for item := range lambda.val {
			if obj, ok := item.(runtime.Object); !ok {
				l.addError(fmt.Errorf("Invalid object type of %#v", obj))
			} else {
				ch <- obj.DeepCopyObject()
			}
		}
	}()
	return l
}
func (lambda *Lambda) LatestCreated() *Lambda {
	l, ch := lambda.clone()
	go func() {
		defer close(ch)
		var latestObj runtime.Object
		var latestTimestamp *time.Time
		for item := range lambda.val {
			item := item
			accessor, _ := meta.Accessor(item)
			if latestTimestamp == nil || accessor.GetCreationTimestamp().Time.After(*latestTimestamp) {
				t := accessor.GetCreationTimestamp().Time
				latestTimestamp = &t
				latestObj = item
			}
		}
		ch <- latestObj
	}()
	return l
}