func (o Observable) Timeout(due time.Duration) Observable {
	observable := Observable(func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var timeout struct {
			sync.Mutex
			at       time.Time
			occurred bool
		}
		timeout.at = subscribeOn.Now().Add(due)
		timer := subscribeOn.ScheduleFutureRecursive(due, func(self func(time.Duration)) {
			if subscriber.Subscribed() {
				timeout.Lock()
				if !timeout.occurred {
					due := timeout.at.Sub(subscribeOn.Now())
					if due > 0 {
						self(due)
					} else {
						timeout.occurred = true
						timeout.Unlock()
						observe(nil, TimeoutOccured, true)
						timeout.Lock()
					}
				}
				timeout.Unlock()
			}
		})
		subscriber.OnUnsubscribe(timer.Cancel)
		observer := func(next interface{}, err error, done bool) {
			if subscriber.Subscribed() {
				timeout.Lock()
				if !timeout.occurred {
					now := subscribeOn.Now()
					if now.Before(timeout.at) {
						timeout.at = now.Add(due)
						timeout.occurred = done
						observe(next, err, done)
					}
				}
				timeout.Unlock()
			}
		}
		o(observer, subscribeOn, subscriber)
	})
	return observable
}
Timeout mirrors the source Observable, but issues an error notification if a
particular period of time elapses without any emitted items.
Timeout schedules a task on the scheduler passed to it during subscription.
package rx
import (
	"sync"
	"time"
	"github.com/reactivego/scheduler"
)
func (o Observable) AuditTime(duration time.Duration) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var audit struct {
			sync.Mutex
			runner scheduler.Runner
			next   interface{}
			done   bool
		}
		auditer := func(self func(time.Duration)) {
			if subscriber.Subscribed() {
				audit.Lock()
				audit.runner = nil
				next := audit.next
				done := audit.done
				audit.Unlock()
				if !done {
					observe(next, nil, false)
				}
			}
		}
		observer := func(next interface{}, err error, done bool) {
			if subscriber.Subscribed() {
				if !done {
					audit.Lock()
					audit.next = next
					if audit.runner == nil {
						audit.runner = subscribeOn.ScheduleFutureRecursive(duration, auditer)
					}
					audit.Unlock()
				} else {
					audit.Lock()
					audit.done = true
					audit.Unlock()
					observe(nil, err, true)
				}
			}
		}
		subscriber.OnUnsubscribe(func() {
			audit.Lock()
			if audit.runner != nil {
				audit.runner.Cancel()
			}
			audit.Unlock()
		})
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
func (o ObservableFoo) AuditTime(duration time.Duration) ObservableFoo {
	return o.AsObservable().AuditTime(duration).AsObservableFoo()
}
func (o Observable) DebounceTime(duration time.Duration) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var debounce struct {
			sync.Mutex
			runner scheduler.Runner
			next   interface{}
			done   bool
		}
		debouncer := func(self func(time.Duration)) {
			if subscriber.Subscribed() {
				debounce.Lock()
				debounce.runner = nil
				next := debounce.next
				done := debounce.done
				debounce.Unlock()
				if !done {
					observe(next, nil, false)
				}
			}
		}
		observer := func(next interface{}, err error, done bool) {
			if subscriber.Subscribed() {
				if !done {
					debounce.Lock()
					debounce.next = next
					if debounce.runner != nil {
						debounce.runner.Cancel()
					}
					debounce.runner = subscribeOn.ScheduleFutureRecursive(duration, debouncer)
					debounce.Unlock()
				} else {
					debounce.Lock()
					debounce.done = true
					debounce.Unlock()
					observe(nil, err, true)
				}
			}
		}
		subscriber.OnUnsubscribe(func() {
			debounce.Lock()
			if debounce.runner != nil {
				debounce.runner.Cancel()
			}
			debounce.Unlock()
		})
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
func (o ObservableFoo) DebounceTime(duration time.Duration) ObservableFoo {
	return o.AsObservable().DebounceTime(duration).AsObservableFoo()
}
func (o Observable) Delay(duration time.Duration) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		type emission struct {
			at   time.Time
			next interface{}
			err  error
			done bool
		}
		var delay struct {
			sync.Mutex
			emissions []emission
		}
		delayer := subscribeOn.ScheduleFutureRecursive(duration, func(self func(time.Duration)) {
			if subscriber.Subscribed() {
				delay.Lock()
				for _, entry := range delay.emissions {
					delay.Unlock()
					due := entry.at.Sub(subscribeOn.Now())
					if due > 0 {
						self(due)
						return
					}
					observe(entry.next, entry.err, entry.done)
					if entry.done || !subscriber.Subscribed() {
						return
					}
					delay.Lock()
					delay.emissions = delay.emissions[1:]
				}
				delay.Unlock()
				self(duration)  
			}
		})
		subscriber.OnUnsubscribe(delayer.Cancel)
		observer := func(next interface{}, err error, done bool) {
			delay.Lock()
			delay.emissions = append(delay.emissions, emission{subscribeOn.Now().Add(duration), next, err, done})
			delay.Unlock()
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
func (o ObservableFoo) Delay(duration time.Duration) ObservableFoo {
	return o.AsObservable().Delay(duration).AsObservableFoo()
}
func IntervalFoo(interval time.Duration) ObservableFoo {
	observable := func(observe FooObserver, subscribeOn Scheduler, subscriber Subscriber) {
		i := 0
		runner := subscribeOn.ScheduleFutureRecursive(interval, func(self func(time.Duration)) {
			if subscriber.Subscribed() {
				observe(foo(i), nil, false)
				i++
				if subscriber.Subscribed() {
					self(interval)
				}
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}
func (o Observable) SampleTime(window time.Duration) Observable {
	observable := Observable(func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var sample struct {
			sync.Mutex
			at   time.Time
			next interface{}
			done bool
		}
		sampler := subscribeOn.ScheduleFutureRecursive(window, func(self func(time.Duration)) {
			if subscriber.Subscribed() {
				sample.Lock()
				if !sample.done {
					begin := subscribeOn.Now().Add(-window)
					if !sample.at.Before(begin) {
						observe(sample.next, nil, false)
					}
					if subscriber.Subscribed() {
						self(window)
					}
				}
				sample.Unlock()
			}
		})
		subscriber.OnUnsubscribe(sampler.Cancel)
		observer := func(next interface{}, err error, done bool) {
			if subscriber.Subscribed() {
				sample.Lock()
				sample.at = subscribeOn.Now()
				sample.next = next
				sample.done = done
				sample.Unlock()
				if done {
					observe(nil, err, true)
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	})
	return observable
}
func (o ObservableFoo) SampleTime(window time.Duration) ObservableFoo {
	return o.AsObservable().SampleTime(window).AsObservableFoo()
}
func (o Observable) ThrottleTime(duration time.Duration) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var deadline time.Time
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if subscribeOn.Now().After(deadline) {
					observe(next, nil, false)
					deadline = subscribeOn.Now().Add(duration)
				}
			} else {
				observe(nil, err, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
func (o ObservableFoo) ThrottleTime(duration time.Duration) ObservableFoo {
	return o.AsObservable().ThrottleTime(duration).AsObservableFoo()
}
type Time = time.Time
func Ticker(initialDelay time.Duration, intervals ...time.Duration) ObservableTime {
	observable := func(observe TimeObserver, subscribeOn Scheduler, subscriber Subscriber) {
		i := 0
		runner := subscribeOn.ScheduleFutureRecursive(initialDelay, func(self func(time.Duration)) {
			if subscriber.Subscribed() {
				if i == 0 || (i > 0 && len(intervals) > 0) {
					observe(subscribeOn.Now(), nil, false)
				}
				if subscriber.Subscribed() {
					if len(intervals) > 0 {
						self(intervals[i%len(intervals)])
					} else {
						if i == 0 {
							self(0)
						} else {
							var zero time.Time
							observe(zero, nil, true)
						}
					}
				}
				i++
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}
const TimeoutOccured = RxError("timeout occured")