func (o ObservableFoo) PublishReplay(bufferCapacity int, windowDuration time.Duration) FooMulticaster {
	factory := func() SubjectFoo {
		return NewReplaySubjectFoo(bufferCapacity, windowDuration)
	}
	return o.Multicast(factory)
}
jig:template Observable<Foo> PublishReplay
jig:needs Observable<Foo> Multicast, NewReplaySubject<Foo>, <Foo>Multicaster
PublishReplay returns a FooMulticaster for a ReplaySubject to an underlying
FooObservable 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 FooObervable terminates with an error, then subscribed observers
will receive that error. After all observers have unsubscribed due to an
error, the FooMulticaster does an internal reset just before the next
observer subscribes.
package rx
import (
	"sync"
	"sync/atomic"
	"time"
	"github.com/reactivego/scheduler"
	"github.com/reactivego/subscriber"
)
type Connectable func(Scheduler, Subscriber)
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
}
type FooMulticaster struct {
	ObservableFoo
	Connectable
}
func (o ObservableFoo) Multicast(factory func() SubjectFoo) FooMulticaster {
	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 foo, err error, done bool) {
		if atomic.CompareAndSwapInt32(&subject.state, active, notifying) {
			if s, ok := subject.Load().(SubjectFoo); ok {
				s.FooObserver(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 FooObserver, 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().(SubjectFoo); ok {
			s.ObservableFoo(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 FooMulticaster{ObservableFoo: observable, Connectable: connectable}
}
func (o ObservableFoo) Publish() FooMulticaster {
	return o.Multicast(NewSubjectFoo)
}