rx.go

// Code generated by jig; DO NOT EDIT.

//go:generate jig

package rx

import (
	"fmt"
	"math"
	"runtime"
	"sync"
	"sync/atomic"
	"time"

	"github.com/reactivego/scheduler"
	"github.com/reactivego/subscriber"
)

//jig:name Subscriber

// Subscriber is an interface that can be passed in when subscribing to an
// Observable. It allows a set of observable subscriptions to be canceled
// from a single subscriber at the root of the subscription tree.
type Subscriber = subscriber.Subscriber

//jig:name NewSubscriber

// NewSubscriber creates a new subscriber.
func NewSubscriber() Subscriber {
	return subscriber.New()
}

//jig:name Scheduler

// Scheduler is used to schedule tasks to support subscribing and observing.
type Scheduler = scheduler.Scheduler

//jig:name GoroutineScheduler

func GoroutineScheduler() Scheduler {
	return scheduler.Goroutine
}

//jig:name MakeTrampolineScheduler

func MakeTrampolineScheduler() Scheduler {
	return scheduler.MakeTrampoline()
}

//jig:name Observer

// Observer is a function that gets called whenever the Observable has
// something to report. The next argument is the item value that is only
// valid when the done argument is false. When done is true and the err
// argument is not nil, then the Observable has terminated with an error.
// When done is true and the err argument is nil, then the Observable has
// completed normally.
type Observer func(next interface{}, err error, done bool)

//jig:name Observable

// Observable is a function taking an Observer, Scheduler and Subscriber.
// Calling it will subscribe the Observer to events from the Observable.
type Observable func(Observer, Scheduler, Subscriber)

//jig:name BoolObserver

// BoolObserver is a function that gets called whenever the Observable has
// something to report. The next argument is the item value that is only
// valid when the done argument is false. When done is true and the err
// argument is not nil, then the Observable has terminated with an error.
// When done is true and the err argument is nil, then the Observable has
// completed normally.
type BoolObserver func(next bool, err error, done bool)

//jig:name ObservableBool

// ObservableBool is a function taking an Observer, Scheduler and Subscriber.
// Calling it will subscribe the Observer to events from the Observable.
type ObservableBool func(BoolObserver, Scheduler, Subscriber)

//jig:name IntObserver

// IntObserver is a function that gets called whenever the Observable has
// something to report. The next argument is the item value that is only
// valid when the done argument is false. When done is true and the err
// argument is not nil, then the Observable has terminated with an error.
// When done is true and the err argument is nil, then the Observable has
// completed normally.
type IntObserver func(next int, err error, done bool)

//jig:name ObservableInt

// ObservableInt is a function taking an Observer, Scheduler and Subscriber.
// Calling it will subscribe the Observer to events from the Observable.
type ObservableInt func(IntObserver, Scheduler, Subscriber)

//jig:name ObservableObserver

// ObservableObserver is a function that gets called whenever the Observable has
// something to report. The next argument is the item value that is only
// valid when the done argument is false. When done is true and the err
// argument is not nil, then the Observable has terminated with an error.
// When done is true and the err argument is nil, then the Observable has
// completed normally.
type ObservableObserver func(next Observable, err error, done bool)

//jig:name ObservableObservable

// ObservableObservable is a function taking an Observer, Scheduler and Subscriber.
// Calling it will subscribe the Observer to events from the Observable.
type ObservableObservable func(ObservableObserver, Scheduler, Subscriber)

//jig:name SliceObserver

// SliceObserver is a function that gets called whenever the Observable has
// something to report. The next argument is the item value that is only
// valid when the done argument is false. When done is true and the err
// argument is not nil, then the Observable has terminated with an error.
// When done is true and the err argument is nil, then the Observable has
// completed normally.
type SliceObserver func(next Slice, err error, done bool)

//jig:name ObservableSlice

// ObservableSlice is a function taking an Observer, Scheduler and Subscriber.
// Calling it will subscribe the Observer to events from the Observable.
type ObservableSlice func(SliceObserver, Scheduler, Subscriber)

//jig:name Error

// Error signals an error condition.
type Error func(error)

//jig:name Complete

// Complete signals that no more data is to be expected.
type Complete func()

//jig:name Canceled

// Canceled returns true when the observer has unsubscribed.
type Canceled func() bool

//jig:name Next

// Next can be called to emit the next value to the IntObserver.
type Next func(interface{})

//jig:name Create

// Create provides a way of creating an Observable from
// scratch by calling observer methods programmatically.
//
// The create function provided to Create will be called once
// to implement the observable. It is provided with a Next, Error,
// Complete and Canceled function that can be called by the code that
// implements the Observable.
func Create(create func(Next, Error, Complete, Canceled)) Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		runner := scheduler.Schedule(func() {
			if !subscriber.Subscribed() {
				return
			}
			n := func(next interface{}) {
				if subscriber.Subscribed() {
					observe(next, nil, false)
				}
			}
			e := func(err error) {
				if subscriber.Subscribed() {
					var zero interface{}
					observe(zero, err, true)
				}
			}
			c := func() {
				if subscriber.Subscribed() {
					var zero interface{}
					observe(zero, nil, true)
				}
			}
			x := func() bool {
				return !subscriber.Subscribed()
			}
			create(n, e, c, x)
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}

//jig:name CreateRecursive

// CreateRecursive provides a way of creating an Observable from
// scratch by calling observer methods programmatically.
//
// The create function provided to CreateRecursive will be called
// repeatedly to implement the observable. It is provided with a Next, Error
// and Complete function that can be called by the code that implements the
// Observable.
func CreateRecursive(create func(Next, Error, Complete)) Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		done := false
		runner := scheduler.ScheduleRecursive(func(self func()) {
			if !subscriber.Subscribed() {
				return
			}
			n := func(next interface{}) {
				if subscriber.Subscribed() {
					observe(next, nil, false)
				}
			}
			e := func(err error) {
				done = true
				if subscriber.Subscribed() {
					var zero interface{}
					observe(zero, err, true)
				}
			}
			c := func() {
				done = true
				if subscriber.Subscribed() {
					var zero interface{}
					observe(zero, nil, true)
				}
			}
			create(n, e, c)
			if !done && subscriber.Subscribed() {
				self()
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}

//jig:name CreateFutureRecursive

// CreateFutureRecursive provides a way of creating an Observable from
// scratch by calling observer methods programmatically.
//
// The create function provided to CreateFutureRecursive will be called
// repeatedly to implement the observable. It is provided with a Next, Error
// and Complete function that can be called by the code that implements the
// Observable.
//
// The timeout passed in determines the time before calling the create
// function. The time.Duration returned by the create function determines how
// long CreateFutureRecursive has to wait before calling the create function
// again.
func CreateFutureRecursive(timeout time.Duration, create func(Next, Error, Complete) time.Duration) Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		done := false
		runner := scheduler.ScheduleFutureRecursive(timeout, func(self func(time.Duration)) {
			if !subscriber.Subscribed() {
				return
			}
			n := func(next interface{}) {
				if subscriber.Subscribed() {
					observe(next, nil, false)
				}
			}
			e := func(err error) {
				done = true
				if subscriber.Subscribed() {
					var zero interface{}
					observe(zero, err, true)
				}
			}
			c := func() {
				done = true
				if subscriber.Subscribed() {
					var zero interface{}
					observe(zero, nil, true)
				}
			}
			timeout = create(n, e, c)
			if !done && subscriber.Subscribed() {
				self(timeout)
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}

//jig:name Defer

// Defer does not create the Observable until the observer subscribes.
// It creates a fresh Observable for each subscribing observer. Use it to
// create observables that maintain separate state per subscription.
func Defer(factory func() Observable) Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		factory()(observe, scheduler, subscriber)
	}
	return observable
}

//jig:name Empty

// Empty creates an Observable that emits no items but terminates normally.
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
}

//jig:name From

// From creates an Observable from multiple interface{} values passed in.
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
}

//jig:name FromChan

// FromChan creates an Observable from a Go channel of interface{}
// values. This allows the code feeding into the channel to send either an error
// or the next value. The feeding code can send nil or more items and then
// closing the channel will be seen as completion. When the feeding code sends
// an error into the channel, it should close the channel immediately to
// indicate termination with error.
func FromChan(ch <-chan interface{}) Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		cancel := make(chan struct{})
		runner := scheduler.ScheduleRecursive(func(self func()) {
			if !subscriber.Subscribed() {
				return
			}
			select {
			case next, ok := <-ch:
				if !subscriber.Subscribed() {
					return
				}
				if ok {
					err, ok := next.(error)
					if !ok {
						observe(next, nil, false)
						if !subscriber.Subscribed() {
							return
						}
						self()
					} else {
						observe(nil, err, true)
					}
				} else {
					observe(nil, nil, true)
				}
			case <-cancel:
				return
			}
		})
		subscriber.OnUnsubscribe(func() {
			runner.Cancel()
			close(cancel)
		})
	}
	return observable
}

//jig:name Of

// Of emits a variable amount of values in a sequence and then emits a
// complete notification.
func Of(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
}

//jig:name Interval

// Interval creates an Observable that emits a sequence of integers spaced
// by a particular time interval. First integer is not emitted immediately, but
// only after the first time interval has passed. The generated code will do a type
// conversion from int to interface{}.
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
}

//jig:name Just

// Just creates an Observable that emits a particular item.
func Just(element interface{}) Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		runner := scheduler.Schedule(func() {
			if subscriber.Subscribed() {
				observe(element, nil, false)
			}
			if subscriber.Subscribed() {
				var zero interface{}
				observe(zero, nil, true)
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}

//jig:name Never

// Never creates an Observable that emits no items and does't terminate.
func Never() Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
	}
	return observable
}

//jig:name Range

// Range creates an Observable that emits a range of sequential int values.
// The generated code will do a type conversion from int to interface{}.
func Range(start, count int) Observable {
	end := start + count
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		i := start
		runner := scheduler.ScheduleRecursive(func(self func()) {
			if subscriber.Subscribed() {
				if i < end {
					observe(interface{}(i), nil, false)
					if subscriber.Subscribed() {
						i++
						self()
					}
				} else {
					var zero interface{}
					observe(zero, nil, true)
				}
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}

//jig:name Start

// Start creates an Observable that emits the return value of a function.
// It is designed to be used with a function that returns a (interface{}, error) tuple.
// If the error is non-nil the returned Observable will be an Observable that
// emits and error, otherwise it will be a single-value Observable of the value.
func Start(f func() (interface{}, error)) Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		done := false
		runner := scheduler.ScheduleRecursive(func(self func()) {
			if subscriber.Subscribed() {
				if !done {
					if next, err := f(); err == nil {
						observe(next, nil, false)
						if subscriber.Subscribed() {
							done = true
							self()
						}
					} else {
						var zero interface{}
						observe(zero, err, true)
					}
				} else {
					var zero interface{}
					observe(zero, nil, true)
				}
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}

//jig:name Throw

// Throw creates an Observable that emits no items and terminates with an
// error.
func Throw(err error) Observable {
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		runner := scheduler.Schedule(func() {
			if subscriber.Subscribed() {
				var zero interface{}
				observe(zero, err, true)
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}

//jig:name RxError

type RxError string

func (e RxError) Error() string	{ return string(e) }

//jig:name Time

type Time = time.Time

//jig:name TimeObserver

// TimeObserver is a function that gets called whenever the Observable has
// something to report. The next argument is the item value that is only
// valid when the done argument is false. When done is true and the err
// argument is not nil, then the Observable has terminated with an error.
// When done is true and the err argument is nil, then the Observable has
// completed normally.
type TimeObserver func(next Time, err error, done bool)

//jig:name ObservableTime

// ObservableTime is a function taking an Observer, Scheduler and Subscriber.
// Calling it will subscribe the Observer to events from the Observable.
type ObservableTime func(TimeObserver, Scheduler, Subscriber)

//jig:name Ticker

// Ticker creates an ObservableTime that emits a sequence of timestamps after
// an initialDelay has passed. Subsequent timestamps are emitted using a
// schedule of intervals passed in. If only the initialDelay is given, Ticker
// will emit only once.
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
}

//jig:name Timer

// Timer creates an Observable that emits a sequence of integers
// (starting at zero) after an initialDelay has passed. Subsequent values are
// emitted using  a schedule of intervals passed in. If only the initialDelay
// is given, Timer will emit only once.
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
}

//jig:name Slice

type Slice = []interface{}

//jig:name ObservableObservable_CombineLatestAll

// CombineLatestAll flattens a higher order observable
// (e.g. ObservableObservable) by subscribing to
// all emitted observables (ie. Observable entries) until the source
// completes. It will then wait for all of the subscribed Observables
// to emit before emitting the first slice. Whenever any of the subscribed
// observables emits, a new slice will be emitted containing all the latest
// value.
func (o ObservableObservable) CombineLatestAll() ObservableSlice {
	observable := func(observe SliceObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observables := []Observable(nil)
		var observers struct {
			sync.Mutex
			assigned	[]bool
			values		[]interface{}
			initialized	int
			active		int
		}
		makeObserver := func(index int) Observer {
			observer := func(next interface{}, err error, done bool) {
				observers.Lock()
				defer observers.Unlock()
				if observers.active > 0 {
					switch {
					case !done:
						if !observers.assigned[index] {
							observers.assigned[index] = true
							observers.initialized++
						}
						observers.values[index] = next
						if observers.initialized == len(observers.values) {
							observe(observers.values, nil, false)
						}
					case err != nil:
						observers.active = 0
						var zero []interface{}
						observe(zero, err, true)
					default:
						if observers.active--; observers.active == 0 {
							var zero []interface{}
							observe(zero, nil, true)
						}
					}
				}
			}
			return observer
		}

		observer := func(next Observable, err error, done bool) {
			switch {
			case !done:
				observables = append(observables, next)
			case err != nil:
				var zero []interface{}
				observe(zero, err, true)
			default:
				subscribeOn.Schedule(func() {
					if subscriber.Subscribed() {
						numObservables := len(observables)
						observers.assigned = make([]bool, numObservables)
						observers.values = make([]interface{}, numObservables)
						observers.active = numObservables
						for i, v := range observables {
							if !subscriber.Subscribed() {
								return
							}
							v(makeObserver(i), subscribeOn, subscriber)
						}
					}
				})
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name CombineLatest

// CombineLatest will subscribe to all Observables. It will then wait for
// all of them to emit before emitting the first slice. Whenever any of the
// subscribed observables emits, a new slice will be emitted containing all
// the latest value.
func CombineLatest(observables ...Observable) ObservableSlice {
	return FromObservable(observables...).CombineLatestAll()
}

//jig:name Concat

// Concat emits the emissions from two or more Observables without interleaving them.
func Concat(observables ...Observable) Observable {
	if len(observables) == 0 {
		return Empty()
	}
	return observables[0].ConcatWith(observables[1:]...)
}

//jig:name Merge

// Merge combines multiple Observables into one by merging their emissions.
// An error from any of the observables will terminate the merged observables.
func Merge(observables ...Observable) Observable {
	if len(observables) == 0 {
		return Empty()
	}
	return observables[0].MergeWith(observables[1:]...)
}

//jig:name MergeDelayError

// MergeDelayError combines multiple Observables into one by merging their emissions.
// Any error will be deferred until all observables terminate.
func MergeDelayError(observables ...Observable) Observable {
	if len(observables) == 0 {
		return Empty()
	}
	return observables[0].MergeDelayErrorWith(observables[1:]...)
}

//jig:name Println

func Println(a ...interface{}) Observer {
	observer := func(next interface{}, err error, done bool) {
		if !done {
			fmt.Println(append(a, next)...)
		}
	}
	return observer
}

//jig:name Subscription

// Subscription is an alias for the subscriber.Subscription interface type.
type Subscription = subscriber.Subscription

//jig:name Connectable

// Connectable provides the Connect method for a Multicaster.
type Connectable func(Scheduler, Subscriber)

// Connect instructs a multicaster to subscribe to its source and begin
// multicasting items to its subscribers. Connect accepts an optional
// scheduler argument.
func (c Connectable) Connect(schedulers ...Scheduler) Subscription {
	subscriber := subscriber.New()
	schedulers = append(schedulers, scheduler.MakeTrampoline())
	if !schedulers[0].IsConcurrent() {
		subscriber.OnWait(schedulers[0].Wait)
	}
	c(schedulers[0], subscriber)
	return subscriber
}

//jig:name Multicaster

// Multicaster is a multicasting connectable observable. One or more
// Observers can subscribe to it simultaneously. It will subscribe to the
// source Observable when Connect is called. After that, every emission
// from the source is multcast to all subscribed Observers.
type Multicaster struct {
	Observable
	Connectable
}

//jig:name FromObservable

// FromObservable creates an ObservableObservable from multiple Observable values passed in.
func FromObservable(slice ...Observable) ObservableObservable {
	observable := func(observe ObservableObserver, 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 Observable
					observe(zero, nil, true)
				}
			}
		})
		subscriber.OnUnsubscribe(runner.Cancel)
	}
	return observable
}

//jig:name Observable_ConcatWith

// ConcatWith emits the emissions from two or more Observables without interleaving them.
func (o Observable) ConcatWith(other ...Observable) Observable {
	if len(other) == 0 {
		return o
	}
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var (
			observables	= append([]Observable{}, other...)
			observer	Observer
		)
		observer = func(next interface{}, err error, done bool) {
			if !done || err != nil {
				observe(next, err, done)
			} else {
				if len(observables) == 0 {
					var zero interface{}
					observe(zero, nil, true)
				} else {
					o := observables[0]
					observables = observables[1:]
					o(observer, subscribeOn, subscriber)
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_MergeWith

// MergeWith combines multiple Observables into one by merging their emissions.
// An error from any of the observables will terminate the merged observables.
func (o Observable) MergeWith(other ...Observable) Observable {
	if len(other) == 0 {
		return o
	}
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var observers struct {
			sync.Mutex
			done	bool
			len	int
		}
		observer := func(next interface{}, err error, done bool) {
			observers.Lock()
			defer observers.Unlock()
			if !observers.done {
				switch {
				case !done:
					observe(next, nil, false)
				case err != nil:
					observers.done = true
					var zero interface{}
					observe(zero, err, true)
				default:
					if observers.len--; observers.len == 0 {
						var zero interface{}
						observe(zero, nil, true)
					}
				}
			}
		}
		observers.len = 1 + len(other)
		o.AutoUnsubscribe()(observer, subscribeOn, subscriber)
		for _, o := range other {
			if subscriber.Subscribed() {
				o.AutoUnsubscribe()(observer, subscribeOn, subscriber)
			}
		}
	}
	return observable
}

//jig:name Observable_MergeDelayErrorWith

// MergeDelayError combines multiple Observables into one by merging their emissions.
// Any error will be deferred until all observables terminate.
func (o Observable) MergeDelayErrorWith(other ...Observable) Observable {
	if len(other) == 0 {
		return o
	}
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var observers struct {
			sync.Mutex
			len	int
			err	error
		}
		observer := func(next interface{}, err error, done bool) {
			observers.Lock()
			defer observers.Unlock()
			if !done {
				observe(next, nil, false)
			} else {
				if err != nil {
					observers.err = err
				}
				if observers.len--; observers.len == 0 {
					var zero interface{}
					observe(zero, observers.err, true)
				}
			}
		}
		observers.len = 1 + len(other)
		o.AutoUnsubscribe()(observer, subscribeOn, subscriber)
		for _, o := range other {
			if subscriber.Subscribed() {
				o.AutoUnsubscribe()(observer, subscribeOn, subscriber)
			}
		}
	}
	return observable
}

//jig:name Observable_CombineLatestWith

// CombineLatestWith will subscribe to its Observable and all other
// Observables passed in. It will then wait for all of the ObservableBars
// to emit before emitting the first slice. Whenever any of the subscribed
// observables emits, a new slice will be emitted containing all the latest
// value.
func (o Observable) CombineLatestWith(other ...Observable) ObservableSlice {
	return FromObservable(append([]Observable{o}, other...)...).CombineLatestAll()
}

//jig:name Observable_CombineLatestMap

// CombinesLatestMap maps every entry emitted by the Observable into an
// Observable, and then subscribe to it, until the source observable
// completes. It will then wait for all of the Observables to emit before
// emitting the first slice. Whenever any of the subscribed observables emits,
// a new slice will be emitted containing all the latest value.
func (o Observable) CombineLatestMap(project func(interface{}) Observable) ObservableSlice {
	return o.MapObservable(project).CombineLatestAll()
}

//jig:name Observable_CombineLatestMapTo

// CombinesLatestMapTo maps every entry emitted by the Observable into a
// single Observable, and then subscribe to it, until the source
// observable completes. It will then wait for all of the Observables
// to emit before emitting the first slice. Whenever any of the subscribed
// observables emits, a new slice will be emitted containing all the latest
// value.
func (o Observable) CombineLatestMapTo(inner Observable) ObservableSlice {
	project := func(interface{}) Observable { return inner }
	return o.MapObservable(project).CombineLatestAll()
}

//jig:name Observable_ConcatMap

// ConcatMap transforms the items emitted by an Observable by applying a
// function to each item and returning an Observable. The stream of
// Observable items is then flattened by concattenating the emissions from
// the observables without interleaving.
func (o Observable) ConcatMap(project func(interface{}) Observable) Observable {
	return o.MapObservable(project).ConcatAll()
}

//jig:name ObservableObservable_ConcatAll

// ConcatAll flattens a higher order observable by concattenating the observables it emits.
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 Observable_SwitchMap

// SwitchMap transforms the items emitted by an Observable by applying a
// function to each item an returning an Observable. In doing so, it behaves much like
// what used to be called FlatMap, except that whenever a new Observable is emitted
// SwitchMap will unsubscribe from the previous Observable and begin emitting items
// from the newly emitted one.
func (o Observable) SwitchMap(project func(interface{}) Observable) Observable {
	return o.MapObservable(project).SwitchAll()
}

//jig:name LinkErrors

const (
	AlreadyDone		= RxError("already done")
	AlreadySubscribed	= RxError("already subscribed")
	AlreadyWaiting		= RxError("already waiting")
	RecursionNotAllowed	= RxError("recursion not allowed")
	StateTransitionFailed	= RxError("state transition faled")
)

//jig:name LinkEnums

// state
const (
	linkUnsubscribed	= iota
	linkSubscribing
	linkIdle
	linkBusy
	linkError	// done:error
	linkCanceled	// externally:canceled
	linkCompleting
	linkComplete	// done:complete
)

// callbackState
const (
	callbackNil	= iota
	settingCallback
	callbackSet
)

// callbackKind
const (
	linkCallbackOnComplete	= iota
	linkCancelOrCompleted
)

//jig:name link

type linkObserver func(*link, interface{}, error, bool)

type link struct {
	observe		linkObserver
	state		int32
	callbackState	int32
	callbackKind	int
	callback	func()
	subscriber	Subscriber
}

func newInitialLink() *link {
	return &link{state: linkCompleting, subscriber: subscriber.New()}
}

func newLink(observe linkObserver, subscriber Subscriber) *link {
	return &link{
		observe:	observe,
		subscriber:	subscriber.Add(),
	}
}

func (o *link) Observe(next interface{}, err error, done bool) error {
	if !atomic.CompareAndSwapInt32(&o.state, linkIdle, linkBusy) {
		if atomic.LoadInt32(&o.state) > linkBusy {
			return AlreadyDone
		}
		return RecursionNotAllowed
	}
	o.observe(o, next, err, done)
	if done {
		if err != nil {
			if !atomic.CompareAndSwapInt32(&o.state, linkBusy, linkError) {
				return StateTransitionFailed
			}
		} else {
			if !atomic.CompareAndSwapInt32(&o.state, linkBusy, linkCompleting) {
				return StateTransitionFailed
			}
		}
	} else {
		if !atomic.CompareAndSwapInt32(&o.state, linkBusy, linkIdle) {
			return StateTransitionFailed
		}
	}
	if atomic.LoadInt32(&o.callbackState) != callbackSet {
		return nil
	}
	if atomic.CompareAndSwapInt32(&o.state, linkCompleting, linkComplete) {
		o.callback()
	}
	if o.callbackKind == linkCancelOrCompleted {
		if atomic.CompareAndSwapInt32(&o.state, linkIdle, linkCanceled) {
			o.callback()
		}
	}
	return nil
}

func (o *link) SubscribeTo(observable Observable, scheduler Scheduler) error {
	if !atomic.CompareAndSwapInt32(&o.state, linkUnsubscribed, linkSubscribing) {
		return AlreadySubscribed
	}
	observer := func(next interface{}, err error, done bool) {
		o.Observe(next, err, done)
	}
	observable(observer, scheduler, o.subscriber)
	if !atomic.CompareAndSwapInt32(&o.state, linkSubscribing, linkIdle) {
		return StateTransitionFailed
	}
	return nil
}

func (o *link) Cancel(callback func()) error {
	if !atomic.CompareAndSwapInt32(&o.callbackState, callbackNil, settingCallback) {
		return AlreadyWaiting
	}
	o.callbackKind = linkCancelOrCompleted
	o.callback = callback
	if !atomic.CompareAndSwapInt32(&o.callbackState, settingCallback, callbackSet) {
		return StateTransitionFailed
	}
	o.subscriber.Unsubscribe()
	if atomic.CompareAndSwapInt32(&o.state, linkCompleting, linkComplete) {
		o.callback()
	}
	if atomic.CompareAndSwapInt32(&o.state, linkIdle, linkCanceled) {
		o.callback()
	}
	return nil
}

func (o *link) OnComplete(callback func()) error {
	if !atomic.CompareAndSwapInt32(&o.callbackState, callbackNil, settingCallback) {
		return AlreadyWaiting
	}
	o.callbackKind = linkCallbackOnComplete
	o.callback = callback
	if !atomic.CompareAndSwapInt32(&o.callbackState, settingCallback, callbackSet) {
		return StateTransitionFailed
	}
	if atomic.CompareAndSwapInt32(&o.state, linkCompleting, linkComplete) {
		o.callback()
	}
	return nil
}

//jig:name ObservableObservable_SwitchAll

// SwitchAll converts an Observable that emits Observables into a single Observable
// that emits the items emitted by the most-recently-emitted of those Observables.
func (o ObservableObservable) SwitchAll() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(link *link, next interface{}, err error, done bool) {
			if !done || err != nil {
				observe(next, err, done)
			} else {
				link.subscriber.Unsubscribe()
			}
		}
		currentLink := newInitialLink()
		var switcherMutex sync.Mutex
		switcherSubscriber := subscriber.Add()
		switcher := func(next Observable, err error, done bool) {
			switch {
			case !done:
				previousLink := currentLink
				func() {
					switcherMutex.Lock()
					defer switcherMutex.Unlock()
					currentLink = newLink(observer, subscriber)
				}()
				previousLink.Cancel(func() {
					switcherMutex.Lock()
					defer switcherMutex.Unlock()
					currentLink.SubscribeTo(next, subscribeOn)
				})
			case err != nil:
				currentLink.Cancel(func() {
					var zero interface{}
					observe(zero, err, true)
				})
				switcherSubscriber.Unsubscribe()
			default:
				currentLink.OnComplete(func() {
					var zero interface{}
					observe(zero, nil, true)
				})
				switcherSubscriber.Unsubscribe()
			}
		}
		o(switcher, subscribeOn, switcherSubscriber)
	}
	return observable
}

//jig:name Observable_MergeMap

// MergeMap transforms the items emitted by an Observable by applying a
// function to each item an returning an Observable. The stream of Observable
// items is then merged into a single stream of  items using the MergeAll operator.
func (o Observable) MergeMap(project func(interface{}) Observable) Observable {
	return o.MapObservable(project).MergeAll()
}

//jig:name ObservableObservable_MergeAll

// MergeAll flattens a higher order observable by merging the observables it emits.
func (o ObservableObservable) MergeAll() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var observers struct {
			sync.Mutex
			done	bool
			len	int32
		}
		observer := func(next interface{}, err error, done bool) {
			observers.Lock()
			defer observers.Unlock()
			if !observers.done {
				switch {
				case !done:
					observe(next, nil, false)
				case err != nil:
					observers.done = true
					var zero interface{}
					observe(zero, err, true)
				default:
					if atomic.AddInt32(&observers.len, -1) == 0 {
						var zero interface{}
						observe(zero, nil, true)
					}
				}
			}
		}
		merger := func(next Observable, err error, done bool) {
			if !done {
				atomic.AddInt32(&observers.len, 1)
				next.AutoUnsubscribe()(observer, subscribeOn, subscriber)
			} else {
				var zero interface{}
				observer(zero, err, true)
			}
		}
		observers.len += 1
		o.AutoUnsubscribe()(merger, subscribeOn, subscriber)
	}
	return observable
}

//jig:name ObservableObservable_WithLatestFromAll

// WithLatestFromAll flattens a higher order observable (e.g. ObservableObservable) by subscribing
// to all emitted observables (ie. Observable entries) until the source completes. It will then wait
// for all of the subscribed Observables to emit before emitting the first slice. The first
// observable that was emitted by the source will be used as the trigger observable. Whenever the
// trigger observable emits, a new slice will be emitted containing all the latest values.
func (o ObservableObservable) WithLatestFromAll() ObservableSlice {
	observable := func(observe SliceObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observables := []Observable(nil)
		var observers struct {
			sync.Mutex
			assigned	[]bool
			values		[]interface{}
			initialized	int
			done		bool
		}
		makeObserver := func(index int) Observer {
			observer := func(next interface{}, err error, done bool) {
				observers.Lock()
				defer observers.Unlock()
				if !observers.done {
					switch {
					case !done:
						if !observers.assigned[index] {
							observers.assigned[index] = true
							observers.initialized++
						}
						observers.values[index] = next
						if index == 0 && observers.initialized == len(observers.values) {
							observe(observers.values, nil, false)
						}
					case err != nil:
						observers.done = true
						var zero []interface{}
						observe(zero, err, true)
					default:
						if index == 0 {
							observers.done = true
							var zero []interface{}
							observe(zero, nil, true)
						}
					}
				}
			}
			return observer
		}

		observer := func(next Observable, err error, done bool) {
			switch {
			case !done:
				observables = append(observables, next)
			case err != nil:
				var zero []interface{}
				observe(zero, err, true)
			default:
				subscribeOn.Schedule(func() {
					if subscriber.Subscribed() {
						numObservables := len(observables)
						observers.assigned = make([]bool, numObservables)
						observers.values = make([]interface{}, numObservables)
						for i, v := range observables {
							if !subscriber.Subscribed() {
								return
							}
							v(makeObserver(i), subscribeOn, subscriber)
						}
					}
				})
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_WithLatestFrom

// WithLatestFrom will subscribe to all Observables and wait for all of them to emit before emitting
// the first slice. The source observable determines the rate at which the values are emitted. The
// idea is that observables that are faster than the source, don't determine the rate at which the
// resulting observable emits. The observables that are combined with the source will be allowed to
// continue emitting but only will have their last emitted value emitted whenever the source emits.
func (o Observable) WithLatestFrom(other ...Observable) ObservableSlice {
	return FromObservable(append([]Observable{o}, other...)...).WithLatestFromAll()
}

//jig:name Observable_All

// All determines whether all items emitted by an Observable meet some
// criteria.
//
// Pass a predicate function to the All operator that accepts an item emitted
// by the source Observable and returns a boolean value based on an
// evaluation of that item. All returns an ObservableBool that emits a single
// boolean value: true if and only if the source Observable terminates
// normally and every item emitted by the source Observable evaluated as
// true according to this predicate; false if any item emitted by the source
// Observable evaluates as false according to this predicate.
func (o Observable) All(predicate func(next interface{}) bool) ObservableBool {
	observable := func(observe BoolObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			switch {
			case !done:
				if !predicate(next) {
					observe(false, nil, false)
					observe(false, nil, true)
				}
			case err != nil:
				observe(false, err, true)
			default:
				observe(true, nil, false)
				observe(false, nil, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name TypecastFailed

// ErrTypecast is delivered to an observer if the generic value cannot be
// typecast to a specific type.
const TypecastFailed = RxError("typecast failed")

//jig:name Observable_AsObservableBool

// AsObservableBool turns an Observable of interface{} into an ObservableBool.
// If during observing a typecast fails, the error ErrTypecastToBool will be
// emitted.
func (o Observable) AsObservableBool() ObservableBool {
	observable := func(observe BoolObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if nextBool, ok := next.(bool); ok {
					observe(nextBool, err, done)
				} else {
					var zero bool
					observe(zero, TypecastFailed, true)
				}
			} else {
				var zero bool
				observe(zero, err, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_AsObservableInt

// AsObservableInt turns an Observable of interface{} into an ObservableInt.
// If during observing a typecast fails, the error ErrTypecastToInt will be
// emitted.
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
}

//jig:name ObservableBool_AsObservable

// AsObservable turns a typed ObservableBool into an Observable of interface{}.
func (o ObservableBool) AsObservable() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next bool, err error, done bool) {
			observe(interface{}(next), err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name ObservableInt_AsObservable

// AsObservable turns a typed ObservableInt into an Observable of interface{}.
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
}

//jig:name Observable_AuditTime

// AuditTime waits until the source emits and then starts a timer. When the
// timer expires, AuditTime will emit the last value received from the source
// during the time period when the timer was active.
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
}

//jig:name ObservableInt_Average

// Average calculates the average of numbers emitted by an ObservableInt and
// emits this average.
func (o ObservableInt) Average() ObservableInt {
	observable := func(observe IntObserver, subscribeOn Scheduler, subscriber Subscriber) {
		var sum, count int
		observer := func(next int, err error, done bool) {
			if !done {
				sum += next
				count++
			} else {
				if count > 0 {
					observe(sum/count, nil, false)
				}
				var zero int
				observe(zero, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Buffer

// Buffer buffers the source Observable values until closingNotifier emits.
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
}

//jig:name Observable_BufferTime

// BufferTime buffers the source Observable values for a specific time period
// and emits those as a slice periodically in time.
func (o Observable) BufferTime(period time.Duration) ObservableSlice {
	return o.Buffer(Interval(period))
}

//jig:name Observable_Catch

// Catch recovers from an error notification by continuing the sequence without
// emitting the error but by switching to the catch Observable to provide
// items.
func (o Observable) Catch(catch Observable) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if err != nil {
				catch(observe, subscribeOn, subscriber)
			} else {
				observe(next, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_CatchError

// CatchError catches errors on the Observable to be handled by returning a
// new Observable or throwing an error. It is passed a selector function
// that takes as arguments err, which is the error, and caught, which is the
// source observable, in case you'd like to "retry" that observable by
// returning it again. Whatever observable is returned by the selector will be
// used to continue the observable chain.
func (o Observable) CatchError(selector func(err error, caught Observable) Observable) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if err != nil {
				selector(err, o)(observe, subscribeOn, subscriber)
			} else {
				observe(next, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Count

// Count counts the number of items emitted by the source Observable and
// emits only this value.
func (o Observable) Count() ObservableInt {
	observable := func(observe IntObserver, subscribeOn Scheduler, subscriber Subscriber) {
		var count int
		observer := func(next interface{}, err error, done bool) {
			if !done {
				count++
			} else {
				observe(count, nil, false)
				observe(0, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_DebounceTime

// DebounceTime only emits the last item of a burst from an Observable if a
// particular timespan has passed without it emitting another item.
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
}

//jig:name Observable_Delay

// Delay shifts an emission from an Observable forward in time by a particular
// amount of time. The relative time intervals between emissions are preserved.
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
}

//jig:name Observable_Distinct

// Distinct suppress duplicate items emitted by an Observable
func (o Observable) Distinct() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		seen := map[interface{}]struct{}{}
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if _, present := seen[next]; present {
					return
				}
				seen[next] = struct{}{}
			}
			observe(next, err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_DistinctUntilChanged

// DistinctUntilChanged only emits when the current value is different from the last.
func (o Observable) DistinctUntilChanged(equal ...func(interface{}, interface{}) bool) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		if len(equal) == 0 {
			equal = append(equal, func(prev, curr interface{}) bool { return prev == curr })
		}
		var seen struct {
			initialized	bool
			value		interface{}
		}
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if seen.initialized && equal[0](seen.value, next) {
					return
				} else {
					seen.initialized = true
					seen.value = next
				}
			}
			observe(next, err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Do

// Do calls a function for each next value passing through the observable.
func (o Observable) Do(f func(next interface{})) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if !done {
				f(next)
			}
			observe(next, err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_DoOnComplete

// DoOnComplete calls a function when the stream completes.
func (o Observable) DoOnComplete(f func()) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if err == nil && done {
				f()
			}
			observe(next, err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_DoOnError

// DoOnError calls a function for any error on the stream.
func (o Observable) DoOnError(f func(err error)) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if err != nil {
				f(err)
			}
			observe(next, err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_ElementAt

// ElementAt emit only item n emitted by an Observable
func (o Observable) ElementAt(n int) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		i := 0
		observer := func(next interface{}, err error, done bool) {
			if done || i == n {
				observe(next, err, done)
			}
			i++
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Filter

// Filter emits only those items from an Observable that pass a predicate test.
func (o Observable) Filter(predicate func(next interface{}) bool) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if done || predicate(next) {
				observe(next, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Finally

// Finally applies a function for any error or completion on the stream.
// This doesn't expose whether this was an error or a completion.
func (o Observable) Finally(f func()) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if done {
				f()
			}
			observe(next, err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_First

// First emits only the first item, or the first item that meets a condition, from an Observable.
func (o Observable) First() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		start := true
		observer := func(next interface{}, err error, done bool) {
			if done || start {
				observe(next, err, done)
			}
			start = false
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_IgnoreCompletion

// IgnoreCompletion only emits items and never completes, neither with Error nor with Complete.
func (o Observable) IgnoreCompletion() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if !done {
				observe(next, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_IgnoreElements

// IgnoreElements does not emit any items from an Observable but mirrors its termination notification.
func (o Observable) IgnoreElements() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if done {
				observe(next, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Last

// Last emits only the last item emitted by an Observable.
func (o Observable) Last() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		have := false
		var last interface{}
		observer := func(next interface{}, err error, done bool) {
			if done {
				if have {
					observe(last, nil, false)
				}
				observe(nil, err, true)
			} else {
				last = next
				have = true
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Map

// Map transforms the items emitted by an Observable by applying a
// function to each item.
func (o Observable) Map(project func(interface{}) interface{}) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			var mapped interface{}
			if !done {
				mapped = project(next)
			}
			observe(mapped, err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name ObservableInt_MapObservable

// MapObservable transforms the items emitted by an ObservableInt by applying a
// function to each item.
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
}

//jig:name Observable_MapTo

// MapTo transforms the items emitted by an Observable. Emitted values
// are mapped to the same value every time.
func (o Observable) MapTo(value interface{}) Observable {
	return o.Map(func(interface{}) interface{} { return value })
}

//jig:name ObservableInt_Max

// Max determines, and emits, the maximum-valued item emitted by an
// ObservableInt.
func (o ObservableInt) Max() ObservableInt {
	observable := func(observe IntObserver, subscribeOn Scheduler, subscriber Subscriber) {
		var started bool
		var max int
		observer := func(next int, err error, done bool) {
			if started {
				if !done {
					if max < next {
						max = next
					}
				} else {
					observe(max, nil, false)
					var zero int
					observe(zero, err, done)
				}
			} else {
				if !done {
					max = next
					started = true
				} else {
					var zero int
					observe(zero, err, done)
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name ObservableInt_Min

// Min determines, and emits, the minimum-valued item emitted by an
// ObservableInt.
func (o ObservableInt) Min() ObservableInt {
	observable := func(observe IntObserver, subscribeOn Scheduler, subscriber Subscriber) {
		var started bool
		var min int
		observer := func(next int, err error, done bool) {
			if started {
				if !done {
					if min > next {
						min = next
					}
				} else {
					observe(min, nil, false)
					var zero int
					observe(zero, err, done)
				}
			} else {
				if !done {
					min = next
					started = true
				} else {
					var zero int
					observe(zero, err, done)
				}

			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_ObserveOn

// ObserveOn specifies a dispatch function to use for delivering values to the observer.
func (o Observable) ObserveOn(dispatch func(task func())) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			task := func() {
				observe(next, err, done)
			}
			dispatch(task)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_OnlyBool

// OnlyBool filters the value stream of an Observable of interface{} and outputs only the
// bool typed values.
func (o Observable) OnlyBool() ObservableBool {
	observable := func(observe BoolObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if nextBool, ok := next.(bool); ok {
					observe(nextBool, err, done)
				}
			} else {
				var zero bool
				observe(zero, err, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_OnlyInt

// OnlyInt filters the value stream of an Observable of interface{} and outputs only the
// int typed values.
func (o Observable) OnlyInt() 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, err, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Reduce

// Reduce applies a reducer function to each item emitted by an Observable
// and the previous reducer result. The operator accepts a seed argument that
// is passed to the reducer for the first item emitted by the Observable.
// Reduce emits only the final value.
func (o Observable) Reduce(reducer func(interface{}, interface{}) interface{}, seed interface{}) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		state := seed
		observer := func(next interface{}, err error, done bool) {
			if !done {
				state = reducer(state, next)
			} else {
				if err == nil {
					observe(state, nil, false)
				}
				var zero interface{}
				observe(zero, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Repeat

// Repeat creates an Observable that emits a sequence of items repeatedly.
func (o Observable) Repeat(count int) Observable {
	if count == 0 {
		return Empty()
	}
	observable := func(observe Observer, scheduler Scheduler, subscriber Subscriber) {
		var repeated int
		var observer Observer
		observer = func(next interface{}, err error, done bool) {
			if !done || err != nil {
				observe(next, err, done)
			} else {
				repeated++
				if repeated < count {
					o(observer, scheduler, subscriber)
				} else {
					observe(nil, nil, true)
				}
			}
		}
		o(observer, scheduler, subscriber)
	}
	return observable
}

//jig:name Observable_Retry

// Retry if a source Observable sends an error notification, resubscribe to
// it in the hopes that it will complete without error. If count is zero or
// negative, the retry count will be effectively infinite. The scheduler
// passed when subscribing is used by Retry to schedule any retry attempt. The
// time between retries is 1 millisecond, so retry frequency is 1 kHz. Any
// SubscribeOn operators should be called after Retry to prevent lockups
// caused by mixing different schedulers in the same subscription for retrying
// and subscribing.
func (o Observable) Retry(count ...int) Observable {
	count = append(count, int(^uint(0)>>1))
	if count[0] <= 0 {
		count = []int{int(^uint(0) >> 1)}
	}
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var retry struct {
			count		int
			observer	Observer
			subscriber	Subscriber
			resubscribe	func()
		}
		retry.count = count[0]
		retry.observer = func(next interface{}, err error, done bool) {
			if err != nil && retry.count > 0 {
				retry.count--
				retry.subscriber.Done(err)
				subscribeOn.ScheduleFuture(1*time.Millisecond, retry.resubscribe)
			} else {
				observe(next, err, done)
			}
		}
		retry.resubscribe = func() {
			if subscriber.Subscribed() {
				retry.subscriber = subscriber.Add()
				if !subscribeOn.IsConcurrent() {
					retry.subscriber.OnWait(subscribeOn.Wait)
					subscriber.OnWait(func() { retry.subscriber.Wait() })
				}
				o(retry.observer, subscribeOn, retry.subscriber)
			}
		}
		retry.resubscribe()
	}
	return observable
}

//jig:name Observable_SampleTime

// SampleTime emits the most recent item emitted by an Observable within periodic time intervals.
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
}

//jig:name Observable_Scan

// Scan applies a accumulator function to each item emitted by an
// Observable and the previous accumulator result. The operator accepts a
// seed argument that is passed to the accumulator for the first item emitted
// by the Observable. Scan emits every value, both intermediate and final.
func (o Observable) Scan(accumulator func(interface{}, interface{}) interface{}, seed interface{}) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		state := seed
		observer := func(next interface{}, err error, done bool) {
			if !done {
				state = accumulator(state, next)
				observe(state, nil, false)
			} else {
				var zero interface{}
				observe(zero, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Serialize

// Serialize forces an Observable to make serialized calls and to be
// well-behaved.
func (o Observable) Serialize() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var observer struct {
			sync.Mutex
			done	bool
		}
		serializer := func(next interface{}, err error, done bool) {
			observer.Lock()
			defer observer.Unlock()
			if !observer.done {
				observer.done = done
				observe(next, err, done)
			}
		}
		o(serializer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name ErrSingle

const DidNotEmitValue = RxError("expected one value, got none")

const EmittedMultipleValues = RxError("expected one value, got multiple")

//jig:name Observable_Single

// Single enforces that the observable sends exactly one data item and then
// completes. If the observable sends no data before completing or sends more
// than 1 item before completing, this is reported as an error to the observer.
func (o Observable) Single() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var (
			count	int
			latest	interface{}
		)
		observer := func(next interface{}, err error, done bool) {
			if count < 2 {
				if done {
					if err != nil {
						observe(nil, err, true)
					} else {
						if count == 1 {
							observe(latest, nil, false)
							observe(nil, nil, true)
						} else {
							observe(nil, DidNotEmitValue, true)
						}
					}
				} else {
					count++
					if count == 1 {
						latest = next
					} else {
						observe(nil, EmittedMultipleValues, true)
					}
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Skip

// Skip suppresses the first n items emitted by an Observable.
func (o Observable) Skip(n int) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		i := 0
		observer := func(next interface{}, err error, done bool) {
			if done || i >= n {
				observe(next, err, done)
			}
			i++
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_SkipLast

// SkipLast suppresses the last n items emitted by an Observable.
func (o Observable) SkipLast(n int) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		read := 0
		write := 0
		n++
		buffer := make([]interface{}, n)
		observer := func(next interface{}, err error, done bool) {
			if done {
				observe(nil, err, true)
			} else {
				buffer[write] = next
				write = (write + 1) % n
				if write == read {
					observe(buffer[read], nil, false)
					read = (read + 1) % n
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_StartWith

// StartWith returns an observable that, at the moment of subscription, will
// synchronously emit all values provided to this operator, then subscribe to
// the source and mirror all of its emissions to subscribers.
func (o Observable) StartWith(values ...interface{}) Observable {
	return From(values...).ConcatWith(o)
}

//jig:name Observable_SubscribeOn

// SubscribeOn specifies the scheduler an Observable should use when it is
// subscribed to.
func (o Observable) SubscribeOn(scheduler Scheduler) Observable {
	observable := func(observe Observer, _ Scheduler, subscriber Subscriber) {
		if scheduler.IsConcurrent() {
			subscriber.OnWait(nil)
		} else {
			subscriber.OnWait(scheduler.Wait)
		}
		o(observe, scheduler, subscriber)
	}
	return observable
}

//jig:name ObservableInt_Sum

// Sum calculates the sum of numbers emitted by an ObservableInt and emits this sum.
func (o ObservableInt) Sum() ObservableInt {
	observable := func(observe IntObserver, subscribeOn Scheduler, subscriber Subscriber) {
		var sum int
		observer := func(next int, err error, done bool) {
			if !done {
				sum += next
			} else {
				observe(sum, nil, false)
				var zero int
				observe(zero, err, done)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Take

// Take emits only the first n items emitted by an 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
}

//jig:name ObservableInt_Take

// Take emits only the first n items emitted by an ObservableInt.
func (o ObservableInt) Take(n int) ObservableInt {
	return o.AsObservable().Take(n).AsObservableInt()
}

//jig:name Observable_TakeLast

// TakeLast emits only the last n items emitted by an Observable.
func (o Observable) TakeLast(n int) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		read := 0
		write := 0
		n++
		buffer := make([]interface{}, n)
		observer := func(next interface{}, err error, done bool) {
			if done {
				for read != write {
					observe(buffer[read], nil, false)
					read = (read + 1) % n
				}
				observe(nil, err, true)
			} else {
				buffer[write] = next
				write = (write + 1) % n
				if write == read {
					read = (read + 1) % n
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_TakeUntil

// TakeUntil emits items emitted by an Observable until another Observable emits an item.
func (o Observable) TakeUntil(other Observable) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		var serializer struct {
			sync.Mutex
			done	bool
		}
		observer := func(next interface{}, err error, done bool) {
			serializer.Lock()
			defer serializer.Unlock()
			if !serializer.done {
				serializer.done = done
				observe(next, err, done)
			}
		}
		triggerSubscriber := subscriber.Add()
		trigger := func(next interface{}, err error, done bool) {
			if !done {
				observer(nil, nil, true)
			} else {
				triggerSubscriber.Unsubscribe()
			}
		}
		other(trigger, subscribeOn, triggerSubscriber)
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_TakeWhile

// TakeWhile mirrors items emitted by an Observable until a specified condition becomes false.
//
// The TakeWhile mirrors the source Observable until such time as some condition you specify
// becomes false, at which point TakeWhile stops mirroring the source Observable and terminates
// its own Observable.
func (o Observable) TakeWhile(condition func(next interface{}) bool) Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if done || condition(next) {
				observe(next, err, done)
			} else {
				observe(nil, nil, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_ThrottleTime

// ThrottleTime emits when the source emits and then starts a timer during which
// all emissions from the source are ignored. After the timer expires, ThrottleTime
// will again emit the next item the source emits, and so on.
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
}

//jig:name TimeInterval

type TimeInterval struct {
	Value		interface{}
	Interval	time.Duration
}

//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.
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_Timeout

// TimeoutOccured is delivered to an observer if the stream times out.
const TimeoutOccured = RxError("timeout occured")

// 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.
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
}

//jig:name Timestamp

type Timestamp struct {
	Value		interface{}
	Timestamp	time.Time
}

//jig:name Observable_Timestamp

// Timestamp attaches a timestamp to each item emitted by an observable
// indicating when it was emitted.
func (o Observable) Timestamp() ObservableTimestamp {
	observable := func(observe TimestampObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if subscriber.Subscribed() {
				if !done {
					observe(Timestamp{next, subscribeOn.Now()}, nil, false)
				} else {
					var zero Timestamp
					observe(zero, err, done)
				}
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_Println

// Println subscribes to the Observable and prints every item to os.Stdout
// while it waits for completion or error. Returns either the error or nil
// when the Observable completed normally.
// Println uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o Observable) Println(a ...interface{}) error {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next interface{}, err error, done bool) {
		if !done {
			fmt.Println(append(a, next)...)
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	return subscriber.Wait()
}

//jig:name ObservableBool_Println

// Println subscribes to the Observable and prints every item to os.Stdout
// while it waits for completion or error. Returns either the error or nil
// when the Observable completed normally.
// Println uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableBool) Println(a ...interface{}) error {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next bool, err error, done bool) {
		if !done {
			fmt.Println(append(a, next)...)
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	return subscriber.Wait()
}

//jig:name ObservableInt_Println

// Println subscribes to the Observable and prints every item to os.Stdout
// while it waits for completion or error. Returns either the error or nil
// when the Observable completed normally.
// Println uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableInt) Println(a ...interface{}) error {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next int, err error, done bool) {
		if !done {
			fmt.Println(append(a, next)...)
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	return subscriber.Wait()
}

//jig:name ObservableSlice_Println

// Println subscribes to the Observable and prints every item to os.Stdout
// while it waits for completion or error. Returns either the error or nil
// when the Observable completed normally.
// Println uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableSlice) Println(a ...interface{}) error {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next Slice, err error, done bool) {
		if !done {
			fmt.Println(append(a, next)...)
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	return subscriber.Wait()
}

//jig:name Observable_Subscribe

// Subscribe operates upon the emissions and notifications from an Observable.
// This method returns a Subscription.
// Subscribe uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o Observable) Subscribe(observe Observer, schedulers ...Scheduler) Subscription {
	subscriber := subscriber.New()
	schedulers = append(schedulers, scheduler.MakeTrampoline())
	observer := func(next interface{}, err error, done bool) {
		if !done {
			observe(next, err, done)
		} else {
			var zero interface{}
			observe(zero, err, true)
			subscriber.Done(err)
		}
	}
	if !schedulers[0].IsConcurrent() {
		subscriber.OnWait(schedulers[0].Wait)
	}
	o(observer, schedulers[0], subscriber)
	return subscriber
}

//jig:name ObservableBool_Subscribe

// Subscribe operates upon the emissions and notifications from an Observable.
// This method returns a Subscription.
// Subscribe uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableBool) Subscribe(observe BoolObserver, schedulers ...Scheduler) Subscription {
	subscriber := subscriber.New()
	schedulers = append(schedulers, scheduler.MakeTrampoline())
	observer := func(next bool, err error, done bool) {
		if !done {
			observe(next, err, done)
		} else {
			var zero bool
			observe(zero, err, true)
			subscriber.Done(err)
		}
	}
	if !schedulers[0].IsConcurrent() {
		subscriber.OnWait(schedulers[0].Wait)
	}
	o(observer, schedulers[0], subscriber)
	return subscriber
}

//jig:name ObservableInt_Subscribe

// Subscribe operates upon the emissions and notifications from an Observable.
// This method returns a Subscription.
// Subscribe uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableInt) Subscribe(observe IntObserver, schedulers ...Scheduler) Subscription {
	subscriber := subscriber.New()
	schedulers = append(schedulers, scheduler.MakeTrampoline())
	observer := func(next int, err error, done bool) {
		if !done {
			observe(next, err, done)
		} else {
			var zero int
			observe(zero, err, true)
			subscriber.Done(err)
		}
	}
	if !schedulers[0].IsConcurrent() {
		subscriber.OnWait(schedulers[0].Wait)
	}
	o(observer, schedulers[0], subscriber)
	return subscriber
}

//jig:name ObservableSlice_Subscribe

// Subscribe operates upon the emissions and notifications from an Observable.
// This method returns a Subscription.
// Subscribe uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableSlice) Subscribe(observe SliceObserver, schedulers ...Scheduler) Subscription {
	subscriber := subscriber.New()
	schedulers = append(schedulers, scheduler.MakeTrampoline())
	observer := func(next Slice, err error, done bool) {
		if !done {
			observe(next, err, done)
		} else {
			var zero Slice
			observe(zero, err, true)
			subscriber.Done(err)
		}
	}
	if !schedulers[0].IsConcurrent() {
		subscriber.OnWait(schedulers[0].Wait)
	}
	o(observer, schedulers[0], subscriber)
	return subscriber
}

//jig:name Observable_ToChan

// ToChan returns a channel that emits interface{} values. If the source
// observable does not emit values but emits an error or complete, then the
// returned channel will enit any error and then close without emitting any
// values.
//
// ToChan uses the public scheduler.Goroutine variable for scheduling, because
// it needs the concurrency so the returned channel can be used by used
// by the calling code directly. To be able to cancel ToChan, you will need to
// create a subscriber yourself and pass it to ToChan as an argument.
func (o Observable) ToChan(subscribers ...Subscriber) <-chan interface{} {
	subscribers = append(subscribers, subscriber.New())
	scheduler := scheduler.Goroutine
	donech := make(chan struct{})
	nextch := make(chan interface{})
	const (
		idle	= iota
		busy
		closed
	)
	state := int32(idle)
	observer := func(next interface{}, err error, done bool) {
		if atomic.CompareAndSwapInt32(&state, idle, busy) {
			if err != nil {
				next = err
			}
			if !done || err != nil {
				select {
				case <-donech:
					atomic.StoreInt32(&state, closed)
				default:
					select {
					case <-donech:
						atomic.StoreInt32(&state, closed)
					case nextch <- next:
					}
				}
			}
			if done {
				atomic.StoreInt32(&state, closed)
				subscribers[0].Done(err)
			}
			if !atomic.CompareAndSwapInt32(&state, busy, idle) {
				close(nextch)
			}
		}
	}
	subscribers[0].OnUnsubscribe(func() {
		close(donech)
		if atomic.CompareAndSwapInt32(&state, busy, closed) {
			return
		}
		if atomic.CompareAndSwapInt32(&state, idle, closed) {
			close(nextch)
			return
		}
	})
	o(observer, scheduler, subscribers[0])
	return nextch
}

//jig:name ObservableBool_ToChan

// ToChan returns a channel that emits bool values. If the source observable does
// not emit values but emits an error or complete, then the returned channel
// will close without emitting any values.
//
// ToChan uses the public scheduler.Goroutine variable for scheduling, because
// it needs the concurrency so the returned channel can be used by used
// by the calling code directly. To be able to cancel ToChan, you will need to
// create a subscriber yourself and pass it to ToChan as an argument.
func (o ObservableBool) ToChan(subscribers ...Subscriber) <-chan bool {
	subscribers = append(subscribers, subscriber.New())
	scheduler := scheduler.Goroutine
	donech := make(chan struct{})
	nextch := make(chan bool)
	const (
		idle	= iota
		busy
		closed
	)
	state := int32(idle)
	observer := func(next bool, err error, done bool) {
		if atomic.CompareAndSwapInt32(&state, idle, busy) {
			if !done {
				select {
				case <-donech:
					atomic.StoreInt32(&state, closed)
				default:
					select {
					case <-donech:
						atomic.StoreInt32(&state, closed)
					case nextch <- next:
					}
				}
			} else {
				atomic.StoreInt32(&state, closed)
				subscribers[0].Done(err)
			}
			if !atomic.CompareAndSwapInt32(&state, busy, idle) {
				close(nextch)
			}
		}
	}
	subscribers[0].OnUnsubscribe(func() {
		close(donech)
		if atomic.CompareAndSwapInt32(&state, busy, closed) {
			return
		}
		if atomic.CompareAndSwapInt32(&state, idle, closed) {
			close(nextch)
			return
		}
	})
	o(observer, scheduler, subscribers[0])
	return nextch
}

//jig:name ObservableInt_ToChan

// ToChan returns a channel that emits int values. If the source observable does
// not emit values but emits an error or complete, then the returned channel
// will close without emitting any values.
//
// ToChan uses the public scheduler.Goroutine variable for scheduling, because
// it needs the concurrency so the returned channel can be used by used
// by the calling code directly. To be able to cancel ToChan, you will need to
// create a subscriber yourself and pass it to ToChan as an argument.
func (o ObservableInt) ToChan(subscribers ...Subscriber) <-chan int {
	subscribers = append(subscribers, subscriber.New())
	scheduler := scheduler.Goroutine
	donech := make(chan struct{})
	nextch := make(chan int)
	const (
		idle	= iota
		busy
		closed
	)
	state := int32(idle)
	observer := func(next int, err error, done bool) {
		if atomic.CompareAndSwapInt32(&state, idle, busy) {
			if !done {
				select {
				case <-donech:
					atomic.StoreInt32(&state, closed)
				default:
					select {
					case <-donech:
						atomic.StoreInt32(&state, closed)
					case nextch <- next:
					}
				}
			} else {
				atomic.StoreInt32(&state, closed)
				subscribers[0].Done(err)
			}
			if !atomic.CompareAndSwapInt32(&state, busy, idle) {
				close(nextch)
			}
		}
	}
	subscribers[0].OnUnsubscribe(func() {
		close(donech)
		if atomic.CompareAndSwapInt32(&state, busy, closed) {
			return
		}
		if atomic.CompareAndSwapInt32(&state, idle, closed) {
			close(nextch)
			return
		}
	})
	o(observer, scheduler, subscribers[0])
	return nextch
}

//jig:name ObservableSlice_ToChan

// ToChan returns a channel that emits Slice values. If the source observable does
// not emit values but emits an error or complete, then the returned channel
// will close without emitting any values.
//
// ToChan uses the public scheduler.Goroutine variable for scheduling, because
// it needs the concurrency so the returned channel can be used by used
// by the calling code directly. To be able to cancel ToChan, you will need to
// create a subscriber yourself and pass it to ToChan as an argument.
func (o ObservableSlice) ToChan(subscribers ...Subscriber) <-chan Slice {
	subscribers = append(subscribers, subscriber.New())
	scheduler := scheduler.Goroutine
	donech := make(chan struct{})
	nextch := make(chan Slice)
	const (
		idle	= iota
		busy
		closed
	)
	state := int32(idle)
	observer := func(next Slice, err error, done bool) {
		if atomic.CompareAndSwapInt32(&state, idle, busy) {
			if !done {
				select {
				case <-donech:
					atomic.StoreInt32(&state, closed)
				default:
					select {
					case <-donech:
						atomic.StoreInt32(&state, closed)
					case nextch <- next:
					}
				}
			} else {
				atomic.StoreInt32(&state, closed)
				subscribers[0].Done(err)
			}
			if !atomic.CompareAndSwapInt32(&state, busy, idle) {
				close(nextch)
			}
		}
	}
	subscribers[0].OnUnsubscribe(func() {
		close(donech)
		if atomic.CompareAndSwapInt32(&state, busy, closed) {
			return
		}
		if atomic.CompareAndSwapInt32(&state, idle, closed) {
			close(nextch)
			return
		}
	})
	o(observer, scheduler, subscribers[0])
	return nextch
}

//jig:name Observable_ToSingle

// ToSingle blocks until the Observable emits exactly one value or an error.
// The value and any error are returned.
// ToSingle uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o Observable) ToSingle() (entry interface{}, err error) {
	o = o.Single()
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next interface{}, err error, done bool) {
		if !done {
			entry = next
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	err = subscriber.Wait()
	return
}

//jig:name ObservableBool_ToSingle

// ToSingle blocks until the ObservableBool emits exactly one value or an error.
// The value and any error are returned.
// ToSingle uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableBool) ToSingle() (entry bool, err error) {
	o = o.Single()
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next bool, err error, done bool) {
		if !done {
			entry = next
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	err = subscriber.Wait()
	return
}

//jig:name ObservableInt_ToSingle

// ToSingle blocks until the ObservableInt emits exactly one value or an error.
// The value and any error are returned.
// ToSingle uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableInt) ToSingle() (entry int, err error) {
	o = o.Single()
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next int, err error, done bool) {
		if !done {
			entry = next
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	err = subscriber.Wait()
	return
}

//jig:name ObservableSlice_ToSingle

// ToSingle blocks until the ObservableSlice emits exactly one value or an error.
// The value and any error are returned.
// ToSingle uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableSlice) ToSingle() (entry Slice, err error) {
	o = o.Single()
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next Slice, err error, done bool) {
		if !done {
			entry = next
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	err = subscriber.Wait()
	return
}

//jig:name Observable_ToSlice

// ToSlice collects all values from the Observable into an slice. The
// complete slice and any error are returned.
// ToSlice uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o Observable) ToSlice() (slice []interface{}, err error) {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next interface{}, err error, done bool) {
		if !done {
			slice = append(slice, next)
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	err = subscriber.Wait()
	return
}

//jig:name ObservableBool_ToSlice

// ToSlice collects all values from the ObservableBool into an slice. The
// complete slice and any error are returned.
// ToSlice uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableBool) ToSlice() (slice []bool, err error) {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next bool, err error, done bool) {
		if !done {
			slice = append(slice, next)
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	err = subscriber.Wait()
	return
}

//jig:name ObservableInt_ToSlice

// ToSlice collects all values from the ObservableInt into an slice. The
// complete slice and any error are returned.
// ToSlice uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableInt) ToSlice() (slice []int, err error) {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next int, err error, done bool) {
		if !done {
			slice = append(slice, next)
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	err = subscriber.Wait()
	return
}

//jig:name ObservableSlice_ToSlice

// ToSlice collects all values from the ObservableSlice into an slice. The
// complete slice and any error are returned.
// ToSlice uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableSlice) ToSlice() (slice []Slice, err error) {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next Slice, err error, done bool) {
		if !done {
			slice = append(slice, next)
		} else {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	err = subscriber.Wait()
	return
}

//jig:name Observable_Wait

// Wait subscribes to the Observable and waits for completion or error.
// Returns either the error or nil when the Observable completed normally.
// Wait uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o Observable) Wait() error {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next interface{}, err error, done bool) {
		if done {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	return subscriber.Wait()
}

//jig:name ObservableBool_Wait

// Wait subscribes to the Observable and waits for completion or error.
// Returns either the error or nil when the Observable completed normally.
// Wait uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableBool) Wait() error {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next bool, err error, done bool) {
		if done {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	return subscriber.Wait()
}

//jig:name ObservableInt_Wait

// Wait subscribes to the Observable and waits for completion or error.
// Returns either the error or nil when the Observable completed normally.
// Wait uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableInt) Wait() error {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next int, err error, done bool) {
		if done {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	return subscriber.Wait()
}

//jig:name ObservableSlice_Wait

// Wait subscribes to the Observable and waits for completion or error.
// Returns either the error or nil when the Observable completed normally.
// Wait uses a trampoline scheduler created with scheduler.MakeTrampoline().
func (o ObservableSlice) Wait() error {
	subscriber := subscriber.New()
	scheduler := scheduler.MakeTrampoline()
	observer := func(next Slice, err error, done bool) {
		if done {
			subscriber.Done(err)
		}
	}
	subscriber.OnWait(scheduler.Wait)
	o(observer, scheduler, subscriber)
	return subscriber.Wait()
}

//jig:name InvalidCount

const InvalidCount = RxError("invalid count")

//jig:name Multicaster_AutoConnect

// AutoConnect makes a Multicaster behave like an ordinary Observable
// that automatically connects the multicaster to its source when the
// specified number of observers have subscribed to it. If the count is less
// than 1 it will return a Throw(InvalidCount). After connecting, when the
// number of subscribed observers eventually drops to 0, AutoConnect will
// cancel the source connection if it hasn't terminated yet. When subsequently
// the next observer subscribes, AutoConnect will connect to the source only
// when it was previously canceled or because the source terminated with an
// error. So it will not reconnect when the source completed succesfully. This
// specific behavior allows for implementing a caching observable that can be
// retried until it succeeds. Another thing to notice is that AutoConnect will
// disconnect an active connection when the number of observers drops to zero.
// The reason for this is that not doing so would leak a task and leave it
// hanging in the scheduler.
func (o Multicaster) AutoConnect(count int) Observable {
	if count < 1 {
		return Throw(InvalidCount)
	}
	var source struct {
		sync.Mutex
		refcount	int32
		subscriber	Subscriber
	}
	observable := func(observe Observer, subscribeOn Scheduler, withSubscriber Subscriber) {
		withSubscriber.OnUnsubscribe(func() {
			source.Lock()
			if atomic.AddInt32(&source.refcount, -1) == 0 {
				if source.subscriber != nil {
					source.subscriber.Unsubscribe()
				}
			}
			source.Unlock()
		})
		o.Observable(observe, subscribeOn, withSubscriber)
		source.Lock()
		if atomic.AddInt32(&source.refcount, 1) == int32(count) {
			if source.subscriber == nil || source.subscriber.Error() != nil {
				source.subscriber = subscriber.New()
				source.Unlock()
				o.Connectable(subscribeOn, source.subscriber)
				source.Lock()
			}
		}
		source.Unlock()
	}
	return observable
}

//jig:name Multicaster_RefCount

// RefCount makes a Multicaster behave like an ordinary Observable. On
// first Subscribe it will call Connect on its Multicaster and when its last
// subscriber is Unsubscribed it will cancel the source connection by calling
// Unsubscribe on the subscription returned by the call to Connect.
func (o Multicaster) RefCount() Observable {
	var source struct {
		sync.Mutex
		refcount	int32
		subscriber	Subscriber
	}
	observable := func(observe Observer, subscribeOn Scheduler, withSubscriber Subscriber) {
		withSubscriber.OnUnsubscribe(func() {
			source.Lock()
			if atomic.AddInt32(&source.refcount, -1) == 0 {
				source.subscriber.Unsubscribe()
			}
			source.Unlock()
		})
		o.Observable(observe, subscribeOn, withSubscriber)
		source.Lock()
		if atomic.AddInt32(&source.refcount, 1) == 1 {
			source.subscriber = subscriber.New()
			source.Unlock()
			o.Connectable(subscribeOn, source.subscriber)
			source.Lock()
		}
		source.Unlock()
	}
	return observable
}

//jig:name Observable_Multicast

// Multicast converts an ordinary observable into a multicasting connectable
// observable or multicaster for short. A multicaster will only start emitting
// values after its Connect method has been called. The factory method passed
// in should return a new Subject that implements the actual multicasting
// behavior.
func (o Observable) Multicast(factory func() Subject) Multicaster {
	const (
		active	int32	= iota
		notifying
		erred
		completed
	)
	var subject struct {
		state	int32
		atomic.Value
		count	int32
	}
	const (
		unsubscribed	int32	= iota
		subscribed
	)
	var source struct {
		sync.Mutex
		state		int32
		subscriber	Subscriber
	}
	subject.Store(factory())
	observer := func(next interface{}, err error, done bool) {
		if atomic.CompareAndSwapInt32(&subject.state, active, notifying) {
			if s, ok := subject.Load().(Subject); ok {
				s.Observer(next, err, done)
			}
			switch {
			case !done:
				atomic.CompareAndSwapInt32(&subject.state, notifying, active)
			case err != nil:
				if atomic.CompareAndSwapInt32(&subject.state, notifying, erred) {
					source.subscriber.Done(err)
				}
			default:
				if atomic.CompareAndSwapInt32(&subject.state, notifying, completed) {
					source.subscriber.Done(nil)
				}
			}
		}
	}
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		if atomic.AddInt32(&subject.count, 1) == 1 {
			if atomic.CompareAndSwapInt32(&subject.state, erred, active) {
				subject.Store(factory())
			}
		}
		if s, ok := subject.Load().(Subject); ok {
			s.Observable(observe, subscribeOn, subscriber)
		}
		subscriber.OnUnsubscribe(func() {
			atomic.AddInt32(&subject.count, -1)
		})
	}
	connectable := func(subscribeOn Scheduler, subscriber Subscriber) {
		source.Lock()
		if atomic.CompareAndSwapInt32(&source.state, unsubscribed, subscribed) {
			source.subscriber = subscriber
			o(observer, subscribeOn, subscriber)
			subscriber.OnUnsubscribe(func() {
				atomic.CompareAndSwapInt32(&source.state, subscribed, unsubscribed)
			})
		} else {
			source.subscriber.OnUnsubscribe(subscriber.Unsubscribe)
			subscriber.OnUnsubscribe(source.subscriber.Unsubscribe)
		}
		source.Unlock()
	}
	return Multicaster{Observable: observable, Connectable: connectable}
}

//jig:name MakeObserverObservable

const OutOfSubscriptions = RxError("out of subscriptions")

// MakeObserverObservable turns an observer into a multicasting and buffering
// observable. Both the observer and the obeservable are returned. These are
// then used as the core of any Subject implementation. The Observer side is
// used to pass items into the buffering multicaster. This then multicasts the
// items to every Observer that subscribes to the returned Observable.
//
//	age     age below which items are kept to replay to a new subscriber.
//	length  length of the item buffer, number of items kept to replay to a new subscriber.
//	[cap]   Capacity of the item buffer, number of items that can be observed before blocking.
//	[scap]  Capacity of the subscription list, max number of simultaneous subscribers.
func MakeObserverObservable(age time.Duration, length int, capacity ...int) (Observer, Observable) {
	const (
		ms	= time.Millisecond
		us	= time.Microsecond
	)

	type subscription struct {
		cursor		uint64
		state		uint64		// active, canceled, closed
		activated	time.Time	// track activity to deterime backoff
		subscribeOn	Scheduler
	}

	// cursor
	const (
		maxuint64 uint64 = math.MaxUint64	// park unused cursor at maxuint64
	)

	// state
	const (
		active	uint64	= iota
		canceled
		closing
		closed
	)

	type subscriptions struct {
		sync.Mutex
		*sync.Cond
		entries	[]subscription
		access	uint32	// unlocked, locked
	}

	// access
	const (
		unlocked	uint32	= iota
		locked
	)

	type item struct {
		Value	interface{}
		At	time.Time
	}

	type buffer struct {
		age	time.Duration
		keep	uint64
		mod	uint64
		size	uint64

		items	[]item
		begin	uint64
		end	uint64
		commit	uint64
		state	uint64	// active, closed

		subscriptions	subscriptions

		err	error
	}

	make := func(age time.Duration, length int, capacity ...int) *buffer {
		if length < 0 {
			length = 0
		}
		keep := uint64(length)

		cap, scap := length, 32
		switch {
		case len(capacity) >= 2:
			cap, scap = capacity[0], capacity[1]
		case len(capacity) == 1:
			cap = capacity[0]
		}
		if cap < length {
			cap = length
		}
		if cap == 0 {
			cap = 1
		}
		size := uint64(1) << uint(math.Ceil(math.Log2(float64(cap))))

		if scap < 1 {
			scap = 1
		}
		buf := &buffer{
			age:	age,
			keep:	keep,
			mod:	size - 1,
			size:	size,

			items:	make([]item, size),
			end:	size,
			subscriptions: subscriptions{
				entries: make([]subscription, 0, scap),
			},
		}
		buf.subscriptions.Cond = sync.NewCond(&buf.subscriptions.Mutex)
		return buf
	}
	buf := make(age, length, capacity...)

	accessSubscriptions := func(access func([]subscription)) bool {
		gosched := false
		for !atomic.CompareAndSwapUint32(&buf.subscriptions.access, unlocked, locked) {
			runtime.Gosched()
			gosched = true
		}
		access(buf.subscriptions.entries)
		atomic.StoreUint32(&buf.subscriptions.access, unlocked)
		return gosched
	}

	send := func(value interface{}) {
		for buf.commit == buf.end {
			full := false
			subscribeOn := Scheduler(nil)
			gosched := accessSubscriptions(func(subscriptions []subscription) {
				slowest := maxuint64
				for i := range subscriptions {
					current := atomic.LoadUint64(&subscriptions[i].cursor)
					if current < slowest {
						slowest = current
						subscribeOn = subscriptions[i].subscribeOn
					}
				}
				end := atomic.LoadUint64(&buf.end)
				if atomic.LoadUint64(&buf.begin) < slowest && slowest <= end {
					if slowest+buf.keep > end {
						slowest = end - buf.keep + 1
					}
					atomic.StoreUint64(&buf.begin, slowest)
					atomic.StoreUint64(&buf.end, slowest+buf.size)
				} else {
					if slowest == maxuint64 {
						atomic.AddUint64(&buf.begin, 1)
						atomic.AddUint64(&buf.end, 1)
					} else {
						full = true
					}
				}
			})
			if full {
				if !gosched {
					if subscribeOn != nil {
						subscribeOn.Gosched()
					} else {
						runtime.Gosched()
					}
				}
				if atomic.LoadUint64(&buf.state) != active {
					return
				}
			}
		}
		buf.items[buf.commit&buf.mod] = item{Value: value, At: time.Now()}
		atomic.AddUint64(&buf.commit, 1)
		buf.subscriptions.Broadcast()
	}

	close := func(err error) {
		if atomic.CompareAndSwapUint64(&buf.state, active, closing) {
			buf.err = err
			if atomic.CompareAndSwapUint64(&buf.state, closing, closed) {
				accessSubscriptions(func(subscriptions []subscription) {
					for i := range subscriptions {
						atomic.CompareAndSwapUint64(&subscriptions[i].state, active, closed)
					}
				})
			}
		}
		buf.subscriptions.Broadcast()
	}

	observer := func(next interface{}, err error, done bool) {
		if atomic.LoadUint64(&buf.state) == active {
			if !done {
				send(next)
			} else {
				close(err)
			}
		}
	}

	appendSubscription := func(subscribeOn Scheduler) (sub *subscription, err error) {
		accessSubscriptions(func([]subscription) {
			cursor := atomic.LoadUint64(&buf.begin)
			s := &buf.subscriptions
			if len(s.entries) < cap(s.entries) {
				s.entries = append(s.entries, subscription{cursor: cursor, subscribeOn: subscribeOn})
				sub = &s.entries[len(s.entries)-1]
				return
			}
			for i := range s.entries {
				sub = &s.entries[i]
				if atomic.CompareAndSwapUint64(&sub.cursor, maxuint64, cursor) {
					sub.subscribeOn = subscribeOn
					return
				}
			}
			sub = nil
			err = OutOfSubscriptions
		})
		return
	}

	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		sub, err := appendSubscription(subscribeOn)
		if err != nil {
			runner := subscribeOn.Schedule(func() {
				if subscriber.Subscribed() {
					observe(nil, err, true)
				}
			})
			subscriber.OnUnsubscribe(runner.Cancel)
			return
		}
		commit := atomic.LoadUint64(&buf.commit)
		if atomic.LoadUint64(&buf.begin)+buf.keep < commit {
			atomic.StoreUint64(&sub.cursor, commit-buf.keep)
		}
		atomic.StoreUint64(&sub.state, atomic.LoadUint64(&buf.state))
		sub.activated = time.Now()

		receiver := subscribeOn.ScheduleFutureRecursive(0, func(self func(time.Duration)) {
			commit := atomic.LoadUint64(&buf.commit)

			if sub.cursor == commit {
				if atomic.CompareAndSwapUint64(&sub.state, canceled, canceled) {

					atomic.StoreUint64(&sub.cursor, maxuint64)
					return
				} else {

					now := time.Now()
					if now.Before(sub.activated.Add(1 * ms)) {

						self(50 * us)
						return
					} else if now.Before(sub.activated.Add(250 * ms)) {
						if atomic.CompareAndSwapUint64(&sub.state, closed, closed) {

							observe(nil, buf.err, true)
							atomic.StoreUint64(&sub.cursor, maxuint64)
							return
						}

						self(500 * us)
						return
					} else {
						if subscribeOn.IsConcurrent() {

							buf.subscriptions.Lock()
							buf.subscriptions.Wait()
							buf.subscriptions.Unlock()
							sub.activated = time.Now()
							self(0)
							return
						} else {

							self(5 * ms)
							return
						}
					}
				}
			}

			if atomic.LoadUint64(&sub.state) == canceled {
				atomic.StoreUint64(&sub.cursor, maxuint64)
				return
			}
			for ; sub.cursor != commit; atomic.AddUint64(&sub.cursor, 1) {
				item := &buf.items[sub.cursor&buf.mod]
				if buf.age == 0 || item.At.IsZero() || time.Since(item.At) < buf.age {
					observe(item.Value, nil, false)
				}
				if atomic.LoadUint64(&sub.state) == canceled {
					atomic.StoreUint64(&sub.cursor, maxuint64)
					return
				}
			}

			sub.activated = time.Now()
			self(0)
		})
		subscriber.OnUnsubscribe(receiver.Cancel)

		subscriber.OnUnsubscribe(func() {
			atomic.CompareAndSwapUint64(&sub.state, active, canceled)
			buf.subscriptions.Broadcast()
		})
	}
	return observer, observable
}

//jig:name Subject

// Subject is a combination of an Observer and Observable.
// Subjects are special because they are the only reactive constructs that
// support multicasting. The items sent to it through its observer side are
// multicasted to multiple clients subscribed to its observable side.
//
// The Subject exposes all methods from the embedded Observer and
// Observable. Use the Observer Next, Error and Complete methods to feed
// data to it. Use the Observable methods to subscribe to it.
//
// After a subject has been terminated by calling either Error or Complete,
// it goes into terminated state. All subsequent calls to its observer side
// will be silently ignored. All subsequent subscriptions to the observable
// side will be handled according to the specific behavior of the subject.
// There are different types of subjects, see the different NewXxxSubject
// functions for more info.
type Subject struct {
	Observer
	Observable
}

// Next is called by an Observable to emit the next interface{} value to the
// Observer.
func (o Observer) Next(next interface{}) {
	o(next, nil, false)
}

// Error is called by an Observable to report an error to the Observer.
func (o Observer) Error(err error) {
	var zero interface{}
	o(zero, err, true)
}

// Complete is called by an Observable to signal that no more data is
// forthcoming to the Observer.
func (o Observer) Complete() {
	var zero interface{}
	o(zero, nil, true)
}

//jig:name Observer_AsObserver

// AsObserver converts an observer of interface{} items to an observer of
// interface{} items.
func (o Observer) AsObserver() Observer {
	observer := func(next interface{}, err error, done bool) {
		o(next, err, done)
	}
	return observer
}

//jig:name Observable_AsObservable

// AsObservable returns the source Observable unchanged.
// This is a special case needed for internal plumbing.
func (o Observable) AsObservable() Observable {
	return o
}

//jig:name NewSubject

// NewSubject creates a new Subject. After the subject is terminated, all
// subsequent subscriptions to the observable side will be terminated
// immediately with either an Error or Complete notification send to the
// subscribing client
//
// Note that this implementation is blocking. When there are subscribers, the
// observable goroutine is blocked until all subscribers have processed the
// next, error or complete notification.
func NewSubject() Subject {
	observer, observable := MakeObserverObservable(0, 0)
	return Subject{observer.AsObserver(), observable.AsObservable()}
}

//jig:name Observable_Publish

// Publish returns a Multicaster for a Subject to an underlying
// Observable and turns the subject into a connnectable observable. A
// Subject emits to an observer only those items that are emitted by the
// underlying Observable subsequent to the time of the observer subscribes.
// When the underlying Obervable terminates with an error, then subscribed
// observers will receive that error. After all observers have unsubscribed
// due to an error, the Multicaster does an internal reset just before the
// next observer subscribes. So this Publish operator is re-connectable,
// unlike the RxJS 5 behavior that isn't. To simulate the RxJS 5 behavior use
// Publish().AutoConnect(1) this will connect on the first subscription but
// will never re-connect.
func (o Observable) Publish() Multicaster {
	return o.Multicast(NewSubject)
}

//jig:name DefaultReplayCapacity

// DefaultReplayCapacity is the default capacity of a replay buffer when
// a bufferCapacity of 0 is passed to the NewReplaySubject function.
const DefaultReplayCapacity = 16383

//jig:name NewReplaySubject

// NewReplaySubject creates a new ReplaySubject. ReplaySubject ensures that
// all observers see the same sequence of emitted items, even if they
// subscribe after. When bufferCapacity argument is 0, then DefaultReplayCapacity is
// used (currently 16380). When windowDuration argument is 0, then entries added
// to the buffer will remain fresh forever.
func NewReplaySubject(bufferCapacity int, windowDuration time.Duration) Subject {
	if bufferCapacity == 0 {
		bufferCapacity = DefaultReplayCapacity
	}
	observer, observable := MakeObserverObservable(windowDuration, bufferCapacity)
	return Subject{observer.AsObserver(), observable.AsObservable()}
}

//jig:name Observable_PublishReplay

// PublishReplay returns a Multicaster for a ReplaySubject to an underlying
// Observable and turns the subject into a connectable observable. A
// ReplaySubject emits to any observer all of the items that were emitted by
// the source observable, regardless of when the observer subscribes. When the
// underlying Obervable terminates with an error, then subscribed observers
// will receive that error. After all observers have unsubscribed due to an
// error, the Multicaster does an internal reset just before the next
// observer subscribes.
func (o Observable) PublishReplay(bufferCapacity int, windowDuration time.Duration) Multicaster {
	factory := func() Subject {
		return NewReplaySubject(bufferCapacity, windowDuration)
	}
	return o.Multicast(factory)
}

//jig:name TimeIntervalObserver

// TimeIntervalObserver is a function that gets called whenever the Observable has
// something to report. The next argument is the item value that is only
// valid when the done argument is false. When done is true and the err
// argument is not nil, then the Observable has terminated with an error.
// When done is true and the err argument is nil, then the Observable has
// completed normally.
type TimeIntervalObserver func(next TimeInterval, err error, done bool)

//jig:name ObservableTimeInterval

// ObservableTimeInterval is a function taking an Observer, Scheduler and Subscriber.
// Calling it will subscribe the Observer to events from the Observable.
type ObservableTimeInterval func(TimeIntervalObserver, Scheduler, Subscriber)

//jig:name TimestampObserver

// TimestampObserver is a function that gets called whenever the Observable has
// something to report. The next argument is the item value that is only
// valid when the done argument is false. When done is true and the err
// argument is not nil, then the Observable has terminated with an error.
// When done is true and the err argument is nil, then the Observable has
// completed normally.
type TimestampObserver func(next Timestamp, err error, done bool)

//jig:name ObservableTimestamp

// ObservableTimestamp is a function taking an Observer, Scheduler and Subscriber.
// Calling it will subscribe the Observer to events from the Observable.
type ObservableTimestamp func(TimestampObserver, Scheduler, Subscriber)

//jig:name Observable_AutoUnsubscribe

// AutoUnsubscribe will automatically unsubscribe from the source when it signals it is done.
// This Operator subscribes to the source Observable using a separate subscriber. When the source
// observable subsequently signals it is done, the separate subscriber will be Unsubscribed.
func (o Observable) AutoUnsubscribe() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		subscriber = subscriber.Add()
		observer := func(next interface{}, err error, done bool) {
			observe(next, err, done)
			if done {
				subscriber.Unsubscribe()
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_MapObservable

// MapObservable transforms the items emitted by an Observable by applying a
// function to each item.
func (o Observable) MapObservable(project func(interface{}) Observable) ObservableObservable {
	observable := func(observe ObservableObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			var mapped Observable
			if !done {
				mapped = project(next)
			}
			observe(mapped, err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name ObservableObservable_AutoUnsubscribeObservable

// AutoUnsubscribe will automatically unsubscribe from the source when it signals it is done.
// This Operator subscribes to the source Observable using a separate subscriber. When the source
// observable subsequently signals it is done, the separate subscriber will be Unsubscribed.
func (o ObservableObservable) AutoUnsubscribe() ObservableObservable {
	observable := func(observe ObservableObserver, subscribeOn Scheduler, subscriber Subscriber) {
		subscriber = subscriber.Add()
		observer := func(next Observable, err error, done bool) {
			observe(next, err, done)
			if done {
				subscriber.Unsubscribe()
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name ObservableBool_Single

// Single enforces that the observableBool sends exactly one data item and then
// completes. If the observable sends no data before completing or sends more
// than 1 item before completing  this reported as an error to the observer.
func (o ObservableBool) Single() ObservableBool {
	return o.AsObservable().Single().AsObservableBool()
}

//jig:name ObservableInt_Single

// Single enforces that the observableInt sends exactly one data item and then
// completes. If the observable sends no data before completing or sends more
// than 1 item before completing  this reported as an error to the observer.
func (o ObservableInt) Single() ObservableInt {
	return o.AsObservable().Single().AsObservableInt()
}

//jig:name ObservableSlice_Single

// Single enforces that the observableSlice sends exactly one data item and then
// completes. If the observable sends no data before completing or sends more
// than 1 item before completing  this reported as an error to the observer.
func (o ObservableSlice) Single() ObservableSlice {
	return o.AsObservable().Single().AsObservableSlice()
}

//jig:name ObservableSlice_AsObservable

// AsObservable turns a typed ObservableSlice into an Observable of interface{}.
func (o ObservableSlice) AsObservable() Observable {
	observable := func(observe Observer, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next Slice, err error, done bool) {
			observe(interface{}(next), err, done)
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}

//jig:name Observable_AsObservableSlice

// AsObservableSlice turns an Observable of interface{} into an ObservableSlice.
// If during observing a typecast fails, the error ErrTypecastToSlice will be
// emitted.
func (o Observable) AsObservableSlice() ObservableSlice {
	observable := func(observe SliceObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if nextSlice, ok := next.(Slice); ok {
					observe(nextSlice, err, done)
				} else {
					var zero Slice
					observe(zero, TypecastFailed, true)
				}
			} else {
				var zero Slice
				observe(zero, err, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}