func (o Observable) TimeInterval() ObservableTimeInterval {
	observable := func(observe TimeIntervalObserver, subscribeOn Scheduler, subscriber Subscriber) {
		begin := subscribeOn.Now()
		observer := func(next interface{}, err error, done bool) {
			if subscriber.Subscribed() {
				if !done {
					now := subscribeOn.Now()
					observe(TimeInterval{next, now.Sub(begin).Round(time.Millisecond)}, nil, false)
					begin = now
				} else {
					var zero TimeInterval
					observe(zero, err, done)
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
jig:name Observable_TimeInterval
TimeInterval intercepts the items from the source Observable and emits in
their place a struct that indicates the amount of time that elapsed between
pairs of emissions.
package TimeInterval
import (
	"fmt"
	"time"
	"github.com/reactivego/scheduler"
	"github.com/reactivego/subscriber"
)
type Scheduler = scheduler.Scheduler
type Subscriber = subscriber.Subscriber
type Observer func(next interface{}, err error, done bool)
type Observable func(Observer, Scheduler, Subscriber)
func Timer(initialDelay time.Duration, intervals ...time.Duration) Observable {
	observable := func(observe Observer, 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(interface{}(i), nil, false)
				}
				if subscriber.Subscribed() {
					if len(intervals) > 0 {
						self(intervals[i%len(intervals)])
					} else {
						if i == 0 {
							self(0)
						} else {
							var zero interface{}
							observe(zero, nil, true)
						}
					}
				}
				i++
			}
		})
		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
}
type TimeInterval struct {
	Value		interface{}
	Interval	time.Duration
}