func (o ObservableObservable) ConcatAll() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var concat struct {
			sync.Mutex
			observables	[]Observable
			observer	Observer
			subscriber	Subscriber
		}
		var source struct {
			observer	ObservableObserver
			subscriber	Subscriber
		}
		concat.observer = func(next interface{}, err error, done bool) {
			concat.Lock()
			if !done || err != nil {
				observe(next, err, done)
			} else {
				if len(concat.observables) == 0 {
					if !source.subscriber.Subscribed() {
						var zero interface{}
						observe(zero, nil, true)
					}
					concat.observables = nil
				} else {
					observable := concat.observables[0]
					concat.observables = concat.observables[1:]
					observable(concat.observer, subscribeOn, subscriber)
				}
			}
			concat.Unlock()
		}
		source.observer = func(next Observable, err error, done bool) {
			if !done {
				concat.Lock()
				initial := concat.observables == nil
				concat.observables = append(concat.observables, next)
				concat.Unlock()
				if initial {
					var zero interface{}
					concat.observer(zero, nil, true)
				}
			} else {
				concat.Lock()
				initial := concat.observables == nil
				source.subscriber.Done(err)
				concat.Unlock()
				if initial || err != nil {
					var zero interface{}
					concat.observer(zero, err, true)
				}
			}
		}
		source.subscriber = subscriber.Add()
		o(source.observer, subscribeOn, source.subscriber)
	}
	return observable
}
jig:name ObservableObservable_ConcatAll
ConcatAll flattens a higher order observable by concattenating the observables it emits.
package BufferTime
import (
	"fmt"
	"sync"
	"time"
	"github.com/reactivego/scheduler"
	"github.com/reactivego/subscriber"
)
type Scheduler = scheduler.Scheduler
type Subscriber = subscriber.Subscriber
type IntObserver func(next int, err error, done bool)
type ObservableInt func(IntObserver, Scheduler, Subscriber)
func TimerInt(initialDelay time.Duration, intervals ...time.Duration) ObservableInt {
	observable := func(observe IntObserver, 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(int(i), nil, false)
				}
				if subscriber.Subscribed() {
					if len(intervals) > 0 {
						self(intervals[i%len(intervals)])
					} else {
						if i == 0 {
							self(0)
						} else {
							var zero int
							observe(zero, nil, true)
						}
					}
				}
				i++
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}
type Observer func(next interface{}, err error, done bool)
type Observable func(Observer, Scheduler, Subscriber)
func From(slice ...interface{}) Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		i := 0
		runner := scheduler.ScheduleRecursive(func(self func()) {
			if subscriber.Subscribed() {
				if i < len(slice) {
					observe(slice[i], nil, false)
					if subscriber.Subscribed() {
						i++
						self()
					}
				} else {
					var zero interface{}
					observe(zero, nil, true)
				}
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}
func Empty() Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		runner := scheduler.Schedule(func() {
			if subscriber.Subscribed() {
				var zero interface{}
				observe(zero, nil, true)
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}
func (o Observable) Take(n int) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		taken := 0
		observer := func(next interface{}, err error, done bool) {
			if taken < n {
				observe(next, err, done)
				if !done {
					taken++
					if taken >= n {
						observe(nil, nil, true)
					}
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
func (o ObservableInt) Take(n int) ObservableInt {
	return o.AsObservable().Take(n).AsObservableInt()
}
func (o ObservableInt) AsObservable() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next int, err error, done bool) {
			observe(interface{}(next), err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
func (o ObservableInt) ConcatMap(project func(int) Observable) Observable {
	return o.MapObservable(project).ConcatAll()
}
type RxError string
func (e RxError) Error() string	{ return string(e) }
const TypecastFailed = RxError("typecast failed")
func (o Observable) AsObservableInt() ObservableInt {
	observable := func(observe IntObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if nextInt, ok := next.(int); ok {
					observe(nextInt, err, done)
				} else {
					var zero int
					observe(zero, TypecastFailed, true)
				}
			} else {
				var zero int
				observe(zero, err, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
func (o ObservableInt) MapObservable(project func(int) Observable) ObservableObservable {
	observable := func(observe ObservableObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next int, err error, done bool) {
			var mapped Observable
			if !done {
				mapped = project(next)
			}
			observe(mapped, err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
func (o Observable) BufferTime(period time.Duration) ObservableSlice {
	return o.Buffer(Interval(period))
}
type ObservableObserver func(next Observable, err error, done bool)
type ObservableObservable func(ObservableObserver, Scheduler, Subscriber)
type SliceObserver func(next Slice, err error, done bool)
type ObservableSlice func(SliceObserver, Scheduler, Subscriber)
func (o Observable) Buffer(closingNotifier Observable) ObservableSlice {
	observable := func(observe SliceObserver, subscribeOn Scheduler, subscriber Subscriber) {
		var serializer struct {
			sync.Mutex
			next	[]interface{}
			done	bool
		}
		notifier := func(next interface{}, err error, done bool) {
			serializer.Lock()
			defer serializer.Unlock()
			if !serializer.done {
				serializer.done = done
				switch {
				case !done:
					observe(serializer.next, nil, false)
					serializer.next = serializer.next[:0]
				case err != nil:
					observe(nil, err, true)
					serializer.next = nil
				default:
					observe(serializer.next, nil, false)
					observe(nil, nil, true)
					serializer.next = nil
				}
			}
		}
		closingNotifier(notifier, subscribeOn, subscriber)
		observer := func(next interface{}, err error, done bool) {
			serializer.Lock()
			defer serializer.Unlock()
			if !serializer.done {
				serializer.done = done
				switch {
				case !done:
					serializer.next = append(serializer.next, next)
				case err != nil:
					observe(nil, err, true)
					serializer.next = nil
				default:
					observe(serializer.next, nil, false)
					observe(nil, nil, true)
					serializer.next = nil
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
func Interval(interval time.Duration) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		i := 0
		runner := subscribeOn.ScheduleFutureRecursive(interval, func(self func(time.Duration)) {
			if subscriber.Subscribed() {
				observe(interface{}(i), nil, false)
				i++
				if subscriber.Subscribed() {
					self(interval)
				}
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}
type Slice = []interface{}