func (f GenericObservableFilterFactory) Bool(parent BoolObservable) BoolObservable {
	return MapBool2BoolObservable(parent, func(observer BoolObserver) MappingBool2BoolFunc {
			gobserver := BoolObserverAsGenericObserver(observer)
			filter := f(gobserver)
			return func(next bool, err error, complete bool, observer BoolObserver) {
				filter(next, err, complete, gobserver)
			}
		},
	)
}
Convert a GenericObservableFilter to a BoolObservable
package gorx
import (
	"errors"
	"time"
	"sync"
	"sync/atomic"
)
var ErrTimeout = errors.New("timeout")
var MaxReplaySize = 16384
type Subscription interface {
	Dispose()
	Disposed() bool
}
type SubscriptionEvents interface {
	OnUnsubscribe(func ())
}
type closedSubscription struct {}
func (closedSubscription) Dispose() {}
func (closedSubscription) Disposed() bool { return true }
var ClosedSubscription Subscription = closedSubscription{}
type LinkedSubscription struct {
	lock sync.Mutex
	unsubscribed bool
	linked Subscription
}
func NewLinkedSubscription() *LinkedSubscription {
	return &LinkedSubscription{}
}
func (l *LinkedSubscription) Link(subscription Subscription) {
	l.lock.Lock()
	defer l.lock.Unlock()
	if l.linked != nil {
		panic("LinkedSubscription is already linked")
	}
	l.linked = subscription
	if l.unsubscribed {
		l.linked.Dispose()
	}
}
func (l *LinkedSubscription) Dispose() {
	l.lock.Lock()
	defer l.lock.Unlock()
	l.unsubscribed = true
	if l.linked != nil {
		l.linked.Dispose()
	}
}
func (l *LinkedSubscription) Disposed() bool {
	l.lock.Lock()
	defer l.lock.Unlock()
	if l.linked != nil {
		return l.linked.Disposed()
	}
	return l.unsubscribed
 }
type ChannelSubscription chan struct{}
func NewChannelSubscription() ChannelSubscription {
	return make(ChannelSubscription)
}
func (c ChannelSubscription) Dispose() {
	defer recover()
	close(c)
}
func (c ChannelSubscription) Disposed() bool {
	select {
	case _, ok := <-c:
		return !ok
	default:
		return false
	}
}
func (c ChannelSubscription) OnUnsubscribe(handler func ()) {
	go func() {
		<-c
		handler()
	}()
}
type GenericSubscription int32
func NewGenericSubscription() Subscription {
	return new(GenericSubscription)
}
func (t *GenericSubscription) Dispose() {
	atomic.StoreInt32((*int32)(t), 1)
}
func (t *GenericSubscription) Disposed() bool {
	return atomic.LoadInt32((*int32)(t)) == 1
}
type CallbackSubscription func()
func (c *CallbackSubscription) Dispose() {
	if *c != nil {
		(*c)()
		*c = nil
	}
}
func (c *CallbackSubscription) Disposed() bool {
	return *c == nil
}
type TerminationObserver interface {
	Error(error)
	Complete()
}
type GenericObserver interface {
	TerminationObserver
	Next(interface{})
}
type GenericObserverFunc struct {
	f            func(interface{}, error, bool)
	Subscription
}
func NewGenericObserverFunc(f func(interface{}, error, bool)) *GenericObserverFunc {
	return &GenericObserverFunc{f, NewGenericSubscription()}
}
func (f *GenericObserverFunc) Next(next interface{}) { f.f(next, nil, false) }
func (f *GenericObserverFunc) Error(err error)       { f.f(nil, err, false) }
func (f *GenericObserverFunc) Complete()             { f.f(nil, nil, true) }
type GenericObservableFilter func(next interface{}, err error, complete bool, observer GenericObserver)
type GenericObservableFilterFactory func(GenericObserver) GenericObservableFilter
func ignoreCompletionFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				break
			default:
				observer.Next(next)
			}
		}
	}
}
func distinctFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		seen := map[interface{}]struct{}{}
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if _, ok := seen[next]; ok {
					return
				}
				seen[next] = struct{}{}
				observer.Next(next)
			}
		}
	}
}
func elementAtFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		i := 0
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if i == n {
					observer.Next(next)
				}
				i++
			}
		}
	}
}
func filterFilter(f func(next interface{}) bool) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if f(next) {
					observer.Next(next)
				}
			}
		}
	}
}
func firstFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		start := true
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if start {
					observer.Next(next)
				}
				start = false
			}
		}
	}
}
func lastFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		have := false
		var last interface{}
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				if have {
					observer.Next(last)
				}
				observer.Error(err)
			case complete:
				if have {
					observer.Next(last)
				}
				observer.Complete()
			default:
				last = next
				have = true
			}
		}
	}
}
func takeLastFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		read := 0
		write := 0
		n++
		buffer := make([]interface{}, n)
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				for read != write {
					observer.Next(buffer[read])
					read = (read + 1) % n
				}
				observer.Error(err)
			case complete:
				for read != write {
					observer.Next(buffer[read])
					read = (read + 1) % n
				}
				observer.Complete()
			default:
				buffer[write] = next
				write = (write + 1) % n
				if write == read {
					read = (read + 1) % n
				}
			}
		}
	}
}
func takeFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		taken := 0
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			if taken >= n {
				return
			}
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				observer.Next(next)
				taken++
				if taken >= n {
					observer.Complete()
				}
			}
		}
	}
}
func ignoreElementsFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			}
		}
	}
}
func skipFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		i := 0
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if i >= n {
					observer.Next(next)
				}
				i++
			}
		}
	}
}
func skipLastFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		read := 0
		write := 0
		n++
		buffer := make([]interface{}, n)
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				buffer[write] = next
				write = (write + 1) % n
				if write == read {
					observer.Next(buffer[read])
					read = (read + 1) % n
				}
			}
		}
	}
}
func oneFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		count := 0
		var value interface{}
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			if count > 1 {
				return
			}
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				if count == 2 {
					observer.Error(errors.New("expected one value"))
				} else {
					observer.Next(value)
					observer.Complete()
				}
			default:
				count++
				if count == 2 {
					observer.Error(errors.New("expected one value"))
					return
				}
				value = next
			}
		}
	}
}
func sampleFilter(window time.Duration) GenericObservableFilterFactory {
	return func(observer GenericObserver) GenericObservableFilter {
		mutex := &sync.Mutex{}
		cancel := make(chan bool, 1)
		var last interface{}
		haveNew := false
		go func() {
			for {
				select {
				case <-time.After(window):
					mutex.Lock()
					if haveNew {
						observer.Next(last)
						haveNew = false
					}
					mutex.Unlock()
				case <-cancel:
					return
				}
			}
		}()
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				cancel <- true
				observer.Error(err)
			case complete:
				cancel <- true
				observer.Complete()
			default:
				mutex.Lock()
				last = next
				haveNew = true
				mutex.Unlock()
			}
		}
	}
}
func debounceFilter(duration time.Duration) GenericObservableFilterFactory {
	return func(observer GenericObserver) GenericObservableFilter {
		errch := make(chan error)
		completech := make(chan bool)
		valuech := make(chan interface{})
		go func() {
			var timeout <-chan time.Time
			var nextValue interface{}
			for {
				select {
				case <-timeout:
					observer.Next(nextValue)
					timeout = nil
				case nextValue = <-valuech:
					timeout = time.After(duration)
				case err := <-errch:
					if timeout != nil {
						observer.Next(nextValue)
					}
					observer.Error(err)
					return
				case <-completech:
					if timeout != nil {
						observer.Next(nextValue)
					}
					observer.Complete()
					return
				}
			}
		}()
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				errch <- err
			case complete:
				completech <- true
			default:
				valuech <- next
			}
		}
	}
}
type timedEntry struct {
	v interface{}
	t time.Time
}
func replayFilter(size int, duration time.Duration) GenericObservableFilterFactory {
	read := 0
	write := 0
	if size == 0 {
		size = MaxReplaySize
	}
	if duration == 0 {
		duration = time.Hour * 24 * 7 * 52
	}
	size++
	buffer := make([]timedEntry, size)
	return func(observer GenericObserver) GenericObservableFilter {
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			now := time.Now()
			switch {
			case err != nil:
				cursor := read
				for cursor != write {
					if buffer[cursor].t.After(now) {
						observer.Next(buffer[cursor].v)
					}
					cursor = (cursor + 1) % size
				}
				observer.Error(err)
			case complete:
				cursor := read
				for cursor != write {
					if buffer[cursor].t.After(now) {
						observer.Next(buffer[cursor].v)
					}
					cursor = (cursor + 1) % size
				}
				observer.Complete()
			default:
				buffer[write] = timedEntry{next, time.Now().Add(duration)}
				write = (write + 1) % size
				if write == read {
					if buffer[read].t.After(now) {
						observer.Next(buffer[read].v)
					}
					read = (read + 1) % size
				}
			}
		}
	}
}
func Range(start, count int) *IntStream {
	end := start + count
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		for i := start; i < end; i++ {
			if subscription.Disposed() {
				return
			}
			observer.Next(i)
		}
		observer.Complete()
		subscription.Dispose()
	})
}
func Interval(interval time.Duration) *IntStream {
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		i := 0
		for {
			time.Sleep(interval)
			if subscription.Disposed() {
				return
			}
			observer.Next(i)
			i++
		}
	})
}
type BoolObserver interface {
	Next(bool)
	TerminationObserver
}
type BoolSubscriber interface {
	Subscription
	BoolObserver
}
type implBoolSubscriber struct {
	Subscription
	BoolObserver
}
func BoolObserverAsGenericObserver(observer BoolObserver) GenericObserver {
	return NewGenericObserverFunc(func(next interface{}, err error, complete bool) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			observer.Next(next.(bool))
		}
	})
}
func GenericObserverAsBoolObserver(observer GenericObserver) BoolObserver {
	return BoolObserverFunc(func(next bool, err error, complete bool) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			observer.Next(next)
		}
	})
}
type BoolObservableFactory func (observer BoolObserver, subscription Subscription)
func (f BoolObservableFactory) Subscribe(observer BoolObserver) Subscription {
	subscription := NewGenericSubscription()
	go f(observer, subscription)
	return subscription
}
func CreateBool(f func (observer BoolObserver, subscription Subscription)) *BoolStream {
	return FromBoolObservable(BoolObservableFactory(f))
}
func RepeatBool(value bool, count int) *BoolStream {
	return CreateBool(func (observer BoolObserver, subscription Subscription) {
		for i := 0; i < count; i++ {
			if subscription.Disposed() {
				return
			}
			observer.Next(value)
		}
		observer.Complete()
	})
}
func StartBool(f func () (bool, error)) *BoolStream {
	return CreateBool(func (observer BoolObserver, subscription Subscription) {
		if v, err := f(); err != nil {
			observer.Error(err)
		} else {
			observer.Next(v)
			observer.Complete()
		}
	})
}
func PassthroughBool(next bool, err error, complete bool, observer BoolObserver) {
	switch {
	case err != nil:
		observer.Error(err)
	case complete:
		observer.Complete()
	default:
		observer.Next(next)
	}
}
var zeroBool = *new(bool)
type BoolObserverFunc func(bool, error, bool)
func (f BoolObserverFunc) Next(next bool) { f(next, nil, false) }
func (f BoolObserverFunc) Error(err error)  { f(zeroBool, err, false) }
func (f BoolObserverFunc) Complete()        { f(zeroBool, nil, true) }
type BoolObservable interface {
	Subscribe(BoolObserver) Subscription
}