func Skip(n int) Node {
	return Ord(func(pipeline *Pipeline, _ NodeKind, size *int) (receiver Receiver, _ Finalizer) {
		switch {
		case n < 0:  
			*size = 0
			pipeline.SetErr(errors.New("skip filter with unknown size"))
			receiver = func(_ int, _ interface{}) interface{} { return nil }
		case n == 0:  
		case (*size < 0) || (*size > n):
			if *size > 0 {
				*size = n
			}
			seen := 0
			receiver = func(_ int, data interface{}) (result interface{}) {
				if seen >= n {
					result = data
					return
				}
				d := reflect.ValueOf(data)
				l := d.Len()
				if (seen + l) > n {
					result = d.Slice(n-seen, l).Interface()
					seen = n
				} else {
					seen += l
				}
				return
			}
		case *size <= n:
			*size = 0
			receiver = func(_ int, _ interface{}) interface{} { return nil }
		}
		return
	})
}
Skip creates an ordered node with a filter that skips the first n elements
from the data batches it passes to the next filter in the pipeline. If n is
negative, no data is passed through, and the error value of the pipeline is
set to a non-nil value.
package pipeline
import (
	"errors"
	"reflect"
	"sync"
	"sync/atomic"
)
func NewNode(kind NodeKind, filters ...Filter) Node {
	switch kind {
	case Ordered, Sequential:
		return &seqnode{kind: kind, filters: filters}
	case Parallel:
		return &parnode{filters: filters}
	default:
		panic("Invalid NodeKind in pipeline.NewNode.")
	}
}
func Identity(_ *Pipeline, _ NodeKind, _ *int) (_ Receiver, _ Finalizer) {
	return
}
func Receive(receive Receiver) Filter {
	return func(_ *Pipeline, _ NodeKind, _ *int) (receiver Receiver, _ Finalizer) {
		receiver = receive
		return
	}
}
func Finalize(finalize Finalizer) Filter {
	return func(_ *Pipeline, _ NodeKind, _ *int) (_ Receiver, finalizer Finalizer) {
		finalizer = finalize
		return
	}
}
func ReceiveAndFinalize(receive Receiver, finalize Finalizer) Filter {
	return func(_ *Pipeline, _ NodeKind, _ *int) (receiver Receiver, finalizer Finalizer) {
		receiver = receive
		finalizer = finalize
		return
	}
}
type Predicate func(data interface{}) bool
func Every(result *bool, cancelWhenKnown bool, predicate Predicate) Filter {
	*result = true
	return func(pipeline *Pipeline, kind NodeKind, _ *int) (receiver Receiver, finalizer Finalizer) {
		switch kind {
		case Parallel:
			res := int32(1)
			receiver = func(_ int, data interface{}) interface{} {
				if !predicate(data) {
					atomic.StoreInt32(&res, 0)
					if cancelWhenKnown {
						pipeline.Cancel()
					}
				}
				return data
			}
			finalizer = func() {
				if atomic.LoadInt32(&res) == 0 {
					*result = false
				}
			}
		default:
			receiver = func(_ int, data interface{}) interface{} {
				if !predicate(data) {
					*result = false
					if cancelWhenKnown {
						pipeline.Cancel()
					}
				}
				return data
			}
		}
		return
	}
}
func NotEvery(result *bool, cancelWhenKnown bool, predicate Predicate) Filter {
	*result = false
	return func(pipeline *Pipeline, kind NodeKind, _ *int) (receiver Receiver, finalizer Finalizer) {
		switch kind {
		case Parallel:
			res := int32(0)
			receiver = func(_ int, data interface{}) interface{} {
				if !predicate(data) {
					atomic.StoreInt32(&res, 1)
					if cancelWhenKnown {
						pipeline.Cancel()
					}
				}
				return data
			}
			finalizer = func() {
				if atomic.LoadInt32(&res) == 1 {
					*result = true
				}
			}
		default:
			receiver = func(_ int, data interface{}) interface{} {
				if !predicate(data) {
					*result = true
					if cancelWhenKnown {
						pipeline.Cancel()
					}
				}
				return data
			}
		}
		return
	}
}
func Some(result *bool, cancelWhenKnown bool, predicate Predicate) Filter {
	*result = false
	return func(pipeline *Pipeline, kind NodeKind, _ *int) (receiver Receiver, finalizer Finalizer) {
		switch kind {
		case Parallel:
			res := int32(0)
			receiver = func(_ int, data interface{}) interface{} {
				if predicate(data) {
					atomic.StoreInt32(&res, 1)
					if cancelWhenKnown {
						pipeline.Cancel()
					}
				}
				return data
			}
			finalizer = func() {
				if atomic.LoadInt32(&res) == 1 {
					*result = true
				}
			}
		default:
			receiver = func(_ int, data interface{}) interface{} {
				if predicate(data) {
					*result = true
					if cancelWhenKnown {
						pipeline.Cancel()
					}
				}
				return data
			}
		}
		return
	}
}
func NotAny(result *bool, cancelWhenKnown bool, predicate Predicate) Filter {
	*result = true
	return func(pipeline *Pipeline, kind NodeKind, _ *int) (receiver Receiver, finalizer Finalizer) {
		switch kind {
		case Parallel:
			res := int32(1)
			receiver = func(_ int, data interface{}) interface{} {
				if predicate(data) {
					atomic.StoreInt32(&res, 0)
					if cancelWhenKnown {
						pipeline.Cancel()
					}
				}
				return data
			}
			finalizer = func() {
				if atomic.LoadInt32(&res) == 0 {
					*result = false
				}
			}
		default:
			receiver = func(_ int, data interface{}) interface{} {
				if predicate(data) {
					*result = false
					if cancelWhenKnown {
						pipeline.Cancel()
					}
				}
				return data
			}
		}
		return
	}
}
func Slice(result interface{}) Filter {
	res := reflect.ValueOf(result).Elem()
	return func(pipeline *Pipeline, kind NodeKind, _ *int) (receiver Receiver, finalizer Finalizer) {
		if (res.Kind() != reflect.Slice) && !res.CanSet() {
			pipeline.SetErr(errors.New("result is not a settable slice in Pipeline.ToSlice"))
			return
		}
		switch kind {
		case Parallel:
			var m sync.Mutex
			receiver = func(_ int, data interface{}) interface{} {
				if data != nil {
					d := reflect.ValueOf(data)
					m.Lock()
					defer m.Unlock()
					res.Set(reflect.AppendSlice(res, d))
				}
				return data
			}
		default:
			receiver = func(_ int, data interface{}) interface{} {
				if data != nil {
					res.Set(reflect.AppendSlice(res, reflect.ValueOf(data)))
				}
				return data
			}
		}
		return
	}
}
func Count(result *int) Filter {
	return func(pipeline *Pipeline, kind NodeKind, size *int) (receiver Receiver, finalizer Finalizer) {
		switch {
		case *size >= 0:
			*result = *size
		case kind == Parallel:
			var res = int64(0)
			receiver = func(_ int, data interface{}) interface{} {
				if data != nil {
					d := reflect.ValueOf(data)
					atomic.AddInt64(&res, int64(d.Len()))
				}
				return data
			}
			finalizer = func() {
				*result = int(atomic.LoadInt64(&res))
			}
		default:
			receiver = func(_ int, data interface{}) interface{} {
				if data != nil {
					d := reflect.ValueOf(data)
					*result += d.Len()
				}
				return data
			}
		}
		return
	}
}
func Limit(limit int, cancelWhenReached bool) Node {
	return Ord(func(pipeline *Pipeline, _ NodeKind, size *int) (receiver Receiver, _ Finalizer) {
		switch {
		case limit < 0:  
		case limit == 0:
			*size = 0
			if cancelWhenReached {
				pipeline.Cancel()
			}
			receiver = func(_ int, _ interface{}) interface{} { return nil }
		case (*size < 0) || (*size > limit):
			if *size > 0 {
				*size = limit
			}
			seen := 0
			receiver = func(_ int, data interface{}) (result interface{}) {
				if seen >= limit {
					return
				}
				d := reflect.ValueOf(data)
				l := d.Len()
				if (seen + l) > limit {
					result = d.Slice(0, limit-seen).Interface()
					seen = limit
				} else {
					result = data
					seen += l
				}
				if cancelWhenReached && (seen == limit) {
					pipeline.Cancel()
				}
				return
			}
		}
		return
	})
}