func (o ObservableString) Subscribe(observe StringObserver, schedulers ...Scheduler) Subscription {
	subscriber := subscriber.New()
	schedulers = append(schedulers, scheduler.MakeTrampoline())
	observer := func(next string, err error, done bool) {
		if !done {
			observe(next, err, done)
		} else {
			var zero string
			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 ObservableString_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().
package Only
import (
	"github.com/reactivego/scheduler"
	"github.com/reactivego/subscriber"
)
type Error func(error)
type Complete func()
type Canceled func() bool
type Next func(interface{})
type Scheduler = scheduler.Scheduler
type Subscriber = subscriber.Subscriber
type Observer func(next interface{}, err error, done bool)
type Observable func(Observer, Scheduler, Subscriber)
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
}
type StringObserver func(next string, err error, done bool)
type ObservableString func(StringObserver, Scheduler, Subscriber)
func (o Observable) OnlyString() ObservableString {
	observable := func(observe StringObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if nextString, ok := next.(string); ok {
					observe(nextString, err, done)
				}
			} else {
				var zero string
				observe(zero, err, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
type SizeObserver func(next Size, err error, done bool)
type ObservableSize func(SizeObserver, Scheduler, Subscriber)
func (o Observable) OnlySize() ObservableSize {
	observable := func(observe SizeObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if nextSize, ok := next.(Size); ok {
					observe(nextSize, err, done)
				}
			} else {
				var zero Size
				observe(zero, err, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
type PointObserver func(next []point, err error, done bool)
type ObservablePoint func(PointObserver, Scheduler, Subscriber)
func (o Observable) OnlyPoint() ObservablePoint {
	observable := func(observe PointObserver, subscribeOn Scheduler, subscriber Subscriber) {
		observer := func(next interface{}, err error, done bool) {
			if !done {
				if nextPoint, ok := next.([]point); ok {
					observe(nextPoint, err, done)
				}
			} else {
				var zero []point
				observe(zero, err, true)
			}
		}
		o(observer, subscribeOn, subscriber)
	}
	return observable
}
type Subscription = subscriber.Subscription