package rx

// NOTE: This file was generated by github.com/alecthomas/gorx/cmd/gorx. Do not modify.

import (
	"errors"
	"io"
	"sync"
	"sync/atomic"
	"time"

	"net/http"
)

// ErrTimeout is delivered to an observer if the stream times out.
var ErrTimeout = errors.New("timeout")

// MaxReplaySize is the maximum size of a replay buffer. Can be modified.
var MaxReplaySize = 16384

// A Subscription to an observable.
type Subscription interface {
	// Unsubscribe from the subscription.
	io.Closer
	// Closed returns true if this subscription has been unsubscribed.
	Closed() bool
}

// A Subscription that is already closed.
type closedSubscription struct{}

func (closedSubscription) Close() error { return nil }
func (closedSubscription) Closed() bool { return true }

// ClosedSubscription always returns true for Closed()
var ClosedSubscription Subscription = closedSubscription{}

// A LinkedSubscription is a link to a (possible) future Subscription.
type LinkedSubscription struct {
	lock         sync.Mutex
	unsubscribed bool
	linked       Subscription
}

func NewLinkedSubscription() *LinkedSubscription {
	return &LinkedSubscription{}
}

func (l *LinkedSubscription) Link(subscription Subscription) {
	l.lock.Lock()
	defer l.lock.Unlock()
	if l.linked != nil {
		panic("LinkedSubscription is already linked")
	}
	l.linked = subscription
	if l.unsubscribed {
		l.linked.Close()
	}
}

func (l *LinkedSubscription) Close() error {
	l.lock.Lock()
	defer l.lock.Unlock()
	l.unsubscribed = true
	if l.linked != nil {
		return l.linked.Close()
	}
	return nil
}

func (l *LinkedSubscription) Closed() bool {
	l.lock.Lock()
	defer l.lock.Unlock()
	if l.linked != nil {
		return l.linked.Closed()
	}
	return l.unsubscribed
}

// ChannelSubscription is implemented with a channel which is closed when
// unsubscribed.
type ChannelSubscription chan struct{}

func NewChannelSubscription() ChannelSubscription {
	return make(ChannelSubscription)
}

func (c ChannelSubscription) Close() error {
	defer recover()
	close(c)
	return nil
}

func (c ChannelSubscription) Closed() bool {
	select {
	case _, ok := <-c:
		return !ok
	default:
		return false
	}
}

// GenericSubscription is implemented with atomic operations.
type GenericSubscription int32

func NewGenericSubscription() Subscription {
	return new(GenericSubscription)
}

func (t *GenericSubscription) Close() error {
	atomic.StoreInt32((*int32)(t), 1)
	return nil
}

func (t *GenericSubscription) Closed() bool {
	return atomic.LoadInt32((*int32)(t)) == 1
}

// TerminationObserver contains functions for observing termination of a stream.
type TerminationObserver interface {
	Error(error)
	Complete()
}

type GenericObserver interface {
	TerminationObserver
	Next(interface{})
}

type GenericObserverFunc struct {
	f func(interface{}, error, bool)
	Subscription
}

func NewGenericObserverFunc(f func(interface{}, error, bool)) *GenericObserverFunc {
	return &GenericObserverFunc{f, NewGenericSubscription()}
}

func (f *GenericObserverFunc) Next(next interface{}) { f.f(next, nil, false) }
func (f *GenericObserverFunc) Error(err error)       { f.f(nil, err, false) }
func (f *GenericObserverFunc) Complete()             { f.f(nil, nil, true) }

type GenericObservableFilter func(next interface{}, err error, complete bool, observer GenericObserver)

// Turtles all the way down!
type GenericObservableFilterFactory func(GenericObserver) GenericObservableFilter

func ignoreCompletionFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				break
			default:
				observer.Next(next)
			}
		}
	}
}

func distinctFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		seen := map[interface{}]struct{}{}
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if _, ok := seen[next]; ok {
					return
				}
				seen[next] = struct{}{}
				observer.Next(next)
			}
		}
	}
}

func elementAtFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		i := 0
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if i == n {
					observer.Next(next)
				}
				i++
			}
		}
	}
}

func filterFilter(f func(next interface{}) bool) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if f(next) {
					observer.Next(next)
				}
			}
		}
	}
}

func firstFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		start := true
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if start {
					observer.Next(next)
				}
				start = false
			}
		}
	}
}

func lastFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		have := false
		var last interface{}
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				if have {
					observer.Next(last)
				}
				observer.Error(err)
			case complete:
				if have {
					observer.Next(last)
				}
				observer.Complete()
			default:
				last = next
				have = true
			}
		}
	}
}

func takeLastFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		read := 0
		write := 0
		n++
		buffer := make([]interface{}, n)
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				for read != write {
					observer.Next(buffer[read])
					read = (read + 1) % n
				}
				observer.Error(err)
			case complete:
				for read != write {
					observer.Next(buffer[read])
					read = (read + 1) % n
				}
				observer.Complete()
			default:
				buffer[write] = next
				write = (write + 1) % n
				if write == read {
					read = (read + 1) % n
				}
			}
		}
	}
}

func takeFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		taken := 0
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			if taken >= n {
				return
			}
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				observer.Next(next)
				taken++
				if taken >= n {
					observer.Complete()
				}
			}
		}
	}
}

func ignoreElementsFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			}
		}
	}
}

func skipFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		i := 0
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				if i >= n {
					observer.Next(next)
				}
				i++
			}
		}
	}
}

func skipLastFilter(n int) GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		read := 0
		write := 0
		n++
		buffer := make([]interface{}, n)
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				observer.Complete()
			default:
				buffer[write] = next
				write = (write + 1) % n
				if write == read {
					observer.Next(buffer[read])
					read = (read + 1) % n
				}
			}
		}
	}
}

func oneFilter() GenericObservableFilterFactory {
	return func(GenericObserver) GenericObservableFilter {
		count := 0
		var value interface{}
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			if count > 1 {
				return
			}
			switch {
			case err != nil:
				observer.Error(err)
			case complete:
				if count == 2 {
					observer.Error(errors.New("expected one value"))
				} else {
					observer.Next(value)
					observer.Complete()
				}
			default:
				count++
				if count == 2 {
					observer.Error(errors.New("expected one value"))
					return
				}
				value = next
			}
		}
	}
}

func sampleFilter(window time.Duration) GenericObservableFilterFactory {
	return func(observer GenericObserver) GenericObservableFilter {
		mutex := &sync.Mutex{}
		cancel := make(chan bool, 1)
		var last interface{}
		haveNew := false
		go func() {
			for {
				select {
				case <-time.After(window):
					mutex.Lock()
					if haveNew {
						observer.Next(last)
						haveNew = false
					}
					mutex.Unlock()
				case <-cancel:
					return
				}
			}
		}()
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				cancel <- true
				observer.Error(err)
			case complete:
				cancel <- true
				observer.Complete()
			default:
				mutex.Lock()
				last = next
				haveNew = true
				mutex.Unlock()
			}
		}
	}
}

func debounceFilter(duration time.Duration) GenericObservableFilterFactory {
	return func(observer GenericObserver) GenericObservableFilter {
		errch := make(chan error)
		completech := make(chan bool)
		valuech := make(chan interface{})
		go func() {
			var timeout <-chan time.Time
			var nextValue interface{}
			for {
				select {
				case <-timeout:
					observer.Next(nextValue)
					timeout = nil
				case nextValue = <-valuech:
					timeout = time.After(duration)
				case err := <-errch:
					if timeout != nil {
						observer.Next(nextValue)
					}
					observer.Error(err)
					return
				case <-completech:
					if timeout != nil {
						observer.Next(nextValue)
					}
					observer.Complete()
					return
				}
			}
		}()
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			switch {
			case err != nil:
				errch <- err
			case complete:
				completech <- true
			default:
				valuech <- next
			}
		}
	}
}

type timedEntry struct {
	v interface{}
	t time.Time
}

func replayFilter(size int, duration time.Duration) GenericObservableFilterFactory {
	read := 0
	write := 0
	if size == 0 {
		size = MaxReplaySize
	}
	if duration == 0 {
		duration = time.Hour * 24 * 7 * 52
	}
	size++
	buffer := make([]timedEntry, size)
	return func(observer GenericObserver) GenericObservableFilter {
		return func(next interface{}, err error, complete bool, observer GenericObserver) {
			now := time.Now()
			switch {
			case err != nil:
				cursor := read
				for cursor != write {
					if buffer[cursor].t.After(now) {
						observer.Next(buffer[cursor].v)
					}
					cursor = (cursor + 1) % size
				}
				observer.Error(err)
			case complete:
				cursor := read
				for cursor != write {
					if buffer[cursor].t.After(now) {
						observer.Next(buffer[cursor].v)
					}
					cursor = (cursor + 1) % size
				}
				observer.Complete()
			default:
				buffer[write] = timedEntry{next, time.Now().Add(duration)}
				write = (write + 1) % size
				if write == read {
					if buffer[read].t.After(now) {
						observer.Next(buffer[read].v)
					}
					read = (read + 1) % size
				}
			}
		}
	}
}

func Range(start, count int) *IntStream {
	end := start + count
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		for i := start; i < end; i++ {
			if subscription.Closed() {
				return
			}
			observer.Next(i)
		}
		observer.Complete()
		subscription.Close()
	})
}

func Interval(interval time.Duration) *IntStream {
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		i := 0
		for {
			time.Sleep(interval)
			if subscription.Closed() {
				return
			}
			observer.Next(i)
			i++
		}
	})
}

type ResponseObserver interface {
	Next(*http.Response)
	TerminationObserver
}

// A ResponseSubscriber represents a subscribed ResponseObserver.
type ResponseSubscriber interface {
	Subscription
	ResponseObserver
}

type implResponseSubscriber struct {
	Subscription
	ResponseObserver
}

func ResponseObserverAsGenericObserver(observer ResponseObserver) GenericObserver {
	return NewGenericObserverFunc(func(next interface{}, err error, complete bool) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			observer.Next(next.(*http.Response))
		}
	})
}

func GenericObserverAsResponseObserver(observer GenericObserver) ResponseObserver {
	return ResponseObserverFunc(func(next *http.Response, err error, complete bool) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			observer.Next(next)
		}
	})
}

type ResponseObservableFactory func(observer ResponseObserver, subscription Subscription)

func (f ResponseObservableFactory) Subscribe(observer ResponseObserver) Subscription {
	subscription := NewGenericSubscription()
	go f(observer, subscription)
	return subscription
}

// CreateResponse calls f(observer, subscription) to produce values for a stream.
func CreateResponse(f func(observer ResponseObserver, subscription Subscription)) *ResponseStream {
	return FromResponseObservable(ResponseObservableFactory(f))
}

// Repeat value count times.
func RepeatResponse(value *http.Response, count int) *ResponseStream {
	return CreateResponse(func(observer ResponseObserver, subscription Subscription) {
		for i := 0; i < count; i++ {
			if subscription.Closed() {
				return
			}
			observer.Next(value)
		}
		observer.Complete()
	})
}

// StartResponse is designed to be used with functions that return a
// (*http.Response, error) tuple.
//
// If the error is non-nil the returned ResponseStream will be that error,
// otherwise it will be a single-value stream of *http.Response.
func StartResponse(f func() (*http.Response, error)) *ResponseStream {
	return CreateResponse(func(observer ResponseObserver, subscription Subscription) {
		if v, err := f(); err != nil {
			observer.Error(err)
		} else {
			observer.Next(v)
			observer.Complete()
		}
	})
}

func PassthroughResponse(next *http.Response, err error, complete bool, observer ResponseObserver) {
	switch {
	case err != nil:
		observer.Error(err)
	case complete:
		observer.Complete()
	default:
		observer.Next(next)
	}
}

var zeroResponse = *new(*http.Response)

type ResponseObserverFunc func(*http.Response, error, bool)

func (f ResponseObserverFunc) Next(next *http.Response) { f(next, nil, false) }
func (f ResponseObserverFunc) Error(err error)          { f(zeroResponse, err, false) }
func (f ResponseObserverFunc) Complete()                { f(zeroResponse, nil, true) }

type ResponseObservable interface {
	Subscribe(ResponseObserver) Subscription
}

// Convert a GenericObservableFilter to a ResponseObservable
func (f GenericObservableFilterFactory) Response(parent ResponseObservable) ResponseObservable {
	return MapResponse2ResponseObservable(parent, func(observer ResponseObserver) MappingResponse2ResponseFunc {
		gobserver := ResponseObserverAsGenericObserver(observer)
		filter := f(gobserver)
		return func(next *http.Response, err error, complete bool, observer ResponseObserver) {
			filter(next, err, complete, gobserver)
		}
	},
	)
}

func NeverResponse() *ResponseStream {
	return CreateResponse(func(observer ResponseObserver, subscription Subscription) {})
}

func EmptyResponse() *ResponseStream {
	return CreateResponse(func(observer ResponseObserver, subscription Subscription) {
		observer.Complete()
	})
}

func ThrowResponse(err error) *ResponseStream {
	return CreateResponse(func(observer ResponseObserver, subscription Subscription) {
		observer.Error(err)
	})
}

func FromResponseArray(array []*http.Response) *ResponseStream {
	return CreateResponse(func(observer ResponseObserver, subscription Subscription) {
		for _, v := range array {
			if subscription.Closed() {
				return
			}
			observer.Next(v)
		}
		observer.Complete()
		subscription.Close()
	})
}

func FromResponses(array ...*http.Response) *ResponseStream {
	return FromResponseArray(array)
}

func JustResponse(element *http.Response) *ResponseStream {
	return FromResponseArray([]*http.Response{element})
}

func MergeResponse(observables ...ResponseObservable) *ResponseStream {
	if len(observables) == 0 {
		return EmptyResponse()
	}
	return (&ResponseStream{observables[0]}).Merge(observables[1:]...)
}

func MergeResponseDelayError(observables ...ResponseObservable) *ResponseStream {
	if len(observables) == 0 {
		return EmptyResponse()
	}
	return (&ResponseStream{observables[0]}).MergeDelayError(observables[1:]...)
}

func FromResponseChannel(ch <-chan *http.Response) *ResponseStream {
	return CreateResponse(func(observer ResponseObserver, subscription Subscription) {
		for v := range ch {
			if subscription.Closed() {
				return
			}
			observer.Next(v)
		}
		observer.Complete()
	})
}

type ResponseStream struct {
	ResponseObservable
}

func FromResponseObservable(observable ResponseObservable) *ResponseStream {
	return &ResponseStream{observable}
}

func (s *ResponseStream) SubscribeFunc(f func(*http.Response, error, bool)) Subscription {
	return s.Subscribe(ResponseObserverFunc(f))
}

func (s *ResponseStream) SubscribeNext(f func(v *http.Response)) Subscription {
	return s.SubscribeFunc(func(next *http.Response, err error, complete bool) {
		if err == nil && !complete {
			f(next)
		}
	})
}

// Distinct removes duplicate elements in the stream.
func (s *ResponseStream) Distinct() *ResponseStream {
	return FromResponseObservable(distinctFilter().Response(s))
}

// ElementAt yields the Nth element of the stream.
func (s *ResponseStream) ElementAt(n int) *ResponseStream {
	return FromResponseObservable(elementAtFilter(n).Response(s))
}

// Filter elements in the stream on a function.
func (s *ResponseStream) Filter(f func(*http.Response) bool) *ResponseStream {
	return FromResponseObservable(filterFilter(func(v interface{}) bool { return f(v.(*http.Response)) }).Response(s))
}

// Last returns just the first element of the stream.
func (s *ResponseStream) First() *ResponseStream {
	return FromResponseObservable(firstFilter().Response(s))
}

// Last returns just the last element of the stream.
func (s *ResponseStream) Last() *ResponseStream {
	return FromResponseObservable(lastFilter().Response(s))
}

// SkipLast skips the first N elements of the stream.
func (s *ResponseStream) Skip(n int) *ResponseStream {
	return FromResponseObservable(skipFilter(n).Response(s))
}

// SkipLast skips the last N elements of the stream.
func (s *ResponseStream) SkipLast(n int) *ResponseStream {
	return FromResponseObservable(skipLastFilter(n).Response(s))
}

// Take returns just the first N elements of the stream.
func (s *ResponseStream) Take(n int) *ResponseStream {
	return FromResponseObservable(takeFilter(n).Response(s))
}

// TakeLast returns just the last N elements of the stream.
func (s *ResponseStream) TakeLast(n int) *ResponseStream {
	return FromResponseObservable(takeLastFilter(n).Response(s))
}

// IgnoreElements ignores elements of the stream and emits only the completion events.
func (s *ResponseStream) IgnoreElements() *ResponseStream {
	return FromResponseObservable(ignoreElementsFilter().Response(s))
}

func (s *ResponseStream) Replay(size int, duration time.Duration) *ResponseStream {
	return FromResponseObservable(replayFilter(size, duration).Response(s))
}

func (s *ResponseStream) Sample(duration time.Duration) *ResponseStream {
	return FromResponseObservable(sampleFilter(duration).Response(s))
}

func (s *ResponseStream) Debounce(duration time.Duration) *ResponseStream {
	return FromResponseObservable(debounceFilter(duration).Response(s))
}

// Wait for completion of the stream and return any error.
func (s *ResponseStream) Wait() error {
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next *http.Response, err error, complete bool) {
		switch {
		case err != nil:
			errch <- err
		case complete:
			errch <- nil
		default:
		}
	})
	return <-errch
}

func MakeResponseSubscriber(observer ResponseObserver) ResponseSubscriber {
	if subscriber, ok := observer.(ResponseSubscriber); ok {
		return subscriber
	}
	return &implResponseSubscriber{NewGenericSubscription(), observer}
}

type concatResponseSubscriber struct {
	observable  int
	observer    ResponseObserver
	observables []ResponseObservable
	Subscription
}

func (c *concatResponseSubscriber) Next(next *http.Response) {
	c.observer.Next(next)
}

func (c *concatResponseSubscriber) Error(err error) {
	c.observer.Error(err)
	c.observable = len(c.observables)
	c.Close()
}

func (c *concatResponseSubscriber) Complete() {
	c.observable++
	if c.observable >= len(c.observables) {
		c.observer.Complete()
		c.Close()
		return
	}
	c.observables[c.observable].Subscribe(c)
}

type concatResponseObservable struct {
	observables []ResponseObservable
}

func (m *concatResponseObservable) Subscribe(observer ResponseObserver) Subscription {
	if len(m.observables) == 0 {
		observer.Complete()
		return ClosedSubscription
	}
	subscriber := &concatResponseSubscriber{
		observer:     observer,
		Subscription: NewGenericSubscription(),
		observables:  m.observables,
	}
	m.observables[0].Subscribe(subscriber)
	return subscriber
}

func (s *ResponseStream) Concat(observables ...ResponseObservable) *ResponseStream {
	return &ResponseStream{&concatResponseObservable{append([]ResponseObservable{s}, observables...)}}
}

type mergeResponseObservable struct {
	delayError  bool
	observables []ResponseObservable
}

func (m *mergeResponseObservable) Subscribe(observer ResponseObserver) Subscription {
	subscription := NewGenericSubscription()
	lock := sync.Mutex{}
	completed := 0
	var firstError error
	relay := func(next *http.Response, err error, complete bool) {
		lock.Lock()
		defer lock.Unlock()
		if completed >= len(m.observables) {
			return
		}

		switch {
		case err != nil:
			if m.delayError {
				firstError = err
				completed++
			} else {
				observer.Error(err)
				completed = len(m.observables)
			}

		case complete:
			completed++
			if completed == len(m.observables) {
				if firstError != nil {
					observer.Error(firstError)
				} else {
					observer.Complete()
				}
			}
		default:
			observer.Next(next)
		}
	}
	for _, observable := range m.observables {
		observable.Subscribe(ResponseObserverFunc(relay))
	}
	return subscription
}

// Merge an arbitrary number of observables with this one.
// An error from any of the observables will terminate the merged stream.
func (s *ResponseStream) Merge(other ...ResponseObservable) *ResponseStream {
	if len(other) == 0 {
		return s
	}
	return &ResponseStream{&mergeResponseObservable{false, append(other, s)}}
}

// Merge an arbitrary number of observables with this one.
// Any error will be deferred until all observables terminate.
func (s *ResponseStream) MergeDelayError(other ...ResponseObservable) *ResponseStream {
	if len(other) == 0 {
		return s
	}
	return &ResponseStream{&mergeResponseObservable{true, append(other, s)}}
}

type catchResponseObservable struct {
	parent ResponseObservable
	catch  ResponseObservable
}

func (r *catchResponseObservable) Subscribe(observer ResponseObserver) Subscription {
	subscription := NewGenericSubscription()
	run := func(next *http.Response, err error, complete bool) {
		switch {
		case err != nil:
			r.catch.Subscribe(observer)
		case complete:
			observer.Complete()
		default:
			observer.Next(next)
		}
	}
	r.parent.Subscribe(ResponseObserverFunc(run))
	return subscription
}

func (s *ResponseStream) Catch(catch ResponseObservable) *ResponseStream {
	return &ResponseStream{&catchResponseObservable{s, catch}}
}

type retryResponseObservable struct {
	observable ResponseObservable
}

type retryResponseObserver struct {
	observable ResponseObservable
	observer   ResponseObserver
}

func (r *retryResponseObserver) retry(next *http.Response, err error, complete bool) {
	switch {
	case err != nil:
		r.observable.Subscribe(ResponseObserverFunc(r.retry))
	case complete:
		r.observer.Complete()
	default:
		r.observer.Next(next)
	}
}

func (r *retryResponseObservable) Subscribe(observer ResponseObserver) Subscription {
	subscription := NewGenericSubscription()
	ro := &retryResponseObserver{r.observable, observer}
	r.observable.Subscribe(ResponseObserverFunc(ro.retry))
	return subscription
}

func (s *ResponseStream) Retry() *ResponseStream {
	return &ResponseStream{&retryResponseObservable{s}}
}

// Do applies a function for each value passing through the stream.
func (s *ResponseStream) Do(f func(next *http.Response)) *ResponseStream {
	return FromResponseObservable(MapResponse2ResponseObserveNext(s, func(next *http.Response) *http.Response {
		f(next)
		return next
	}))
}

// DoOnError applies a function for any error on the stream.
func (s *ResponseStream) DoOnError(f func(err error)) *ResponseStream {
	return FromResponseObservable(MapResponse2ResponseObserveDirect(s, func(next *http.Response, err error, complete bool, observer ResponseObserver) {
		if err != nil {
			f(err)
		}
		PassthroughResponse(next, err, complete, observer)
	}))
}

// DoOnComplete applies a function when the stream completes.
func (s *ResponseStream) DoOnComplete(f func()) *ResponseStream {
	return FromResponseObservable(MapResponse2ResponseObserveDirect(s, func(next *http.Response, err error, complete bool, observer ResponseObserver) {
		if complete {
			f()
		}
		PassthroughResponse(next, err, complete, observer)
	}))
}

func (s *ResponseStream) Reduce(initial *http.Response, reducer func(*http.Response, *http.Response) *http.Response) *ResponseStream {
	value := initial
	return FromResponseObservable(MapResponse2ResponseObserveDirect(s, func(next *http.Response, err error, complete bool, observer ResponseObserver) {
		switch {
		case err != nil:
			observer.Next(value)
			observer.Error(err)
		case complete:
			observer.Next(value)
			observer.Complete()
		default:
			value = reducer(value, next)
		}
	}))
}

func (s *ResponseStream) Scan(initial *http.Response, f func(*http.Response, *http.Response) *http.Response) *ResponseStream {
	value := initial
	return FromResponseObservable(MapResponse2ResponseObserveDirect(s, func(next *http.Response, err error, complete bool, observer ResponseObserver) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			value = f(value, next)
			observer.Next(value)
		}
	}))
}

type timeoutResponse struct {
	parent  ResponseObservable
	timeout time.Duration
}

func (t *timeoutResponse) Subscribe(observer ResponseObserver) Subscription {
	subscription := NewChannelSubscription()
	cancel := t.parent.Subscribe(observer)
	go func() {
		select {
		case <-time.After(t.timeout):
			observer.Error(ErrTimeout)
			_ = cancel.Close()
			_ = subscription.Close()
		case <-subscription:
			_ = cancel.Close()
		}
	}()
	return subscription
}

func (s *ResponseStream) Timeout(timeout time.Duration) *ResponseStream {
	return &ResponseStream{&timeoutResponse{s, timeout}}

}

// ToOneWithError blocks until the stream emits exactly one value. Otherwise, it errors.
func (s *ResponseStream) ToOneWithError() (*http.Response, error) {
	valuech := make(chan *http.Response, 1)
	errch := make(chan error, 1)
	FromResponseObservable(oneFilter().Response(s)).SubscribeFunc(func(next *http.Response, err error, complete bool) {
		if err != nil {
			errch <- err
		} else if !complete {
			valuech <- next
		}
	})
	select {
	case value := <-valuech:
		return value, nil
	case err := <-errch:
		return zeroResponse, err
	}
}

// ToOne blocks and returns the only value emitted by the stream, or the zero
// value if an error occurs.
func (s *ResponseStream) ToOne() *http.Response {
	value, _ := s.ToOneWithError()
	return value
}

// ToArrayWithError collects all values from the stream into an array,
// returning it and any error.
func (s *ResponseStream) ToArrayWithError() ([]*http.Response, error) {
	array := []*http.Response{}
	completech := make(chan bool, 1)
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next *http.Response, err error, complete bool) {
		switch {
		case err != nil:
			errch <- err
		case complete:
			completech <- true
		default:
			array = append(array, next)
		}
	})
	select {
	case <-completech:
		return array, nil
	case err := <-errch:
		return array, err
	}
}

// ToArray blocks and returns the values from the stream in an array.
func (s *ResponseStream) ToArray() []*http.Response {
	out, _ := s.ToArrayWithError()
	return out
}

// ToChannelWithError returns value and error channels corresponding to the stream elements and any error.
func (s *ResponseStream) ToChannelWithError() (<-chan *http.Response, <-chan error) {
	ch := make(chan *http.Response, 1)
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next *http.Response, err error, complete bool) {
		switch {
		case err != nil:
			errch <- err
			close(errch)
			close(ch)
		case complete:
			close(ch)
		default:
			ch <- next
		}
	})
	return ch, errch
}

func (s *ResponseStream) ToChannel() <-chan *http.Response {
	ch, _ := s.ToChannelWithError()
	return ch
}

// Count returns an IntStream with the count of elements in this stream.
func (s *ResponseStream) Count() *IntStream {
	count := 0
	return FromIntObservable(MapResponse2IntObserveDirect(s, func(next *http.Response, err error, complete bool, observer IntObserver) {
		switch {
		case err != nil:
			observer.Next(count)
			observer.Error(err)
		case complete:
			observer.Next(count)
			observer.Complete()
		default:
			count++
		}
	}))
}

type MappingResponse2ResponseFunc func(next *http.Response, err error, complete bool, observer ResponseObserver)
type MappingResponse2ResponseFuncFactory func(observer ResponseObserver) MappingResponse2ResponseFunc

type MappingResponse2ResponseObservable struct {
	parent ResponseObservable
	mapper MappingResponse2ResponseFuncFactory
}

func (f *MappingResponse2ResponseObservable) Subscribe(observer ResponseObserver) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe(ResponseObserverFunc(func(next *http.Response, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func MapResponse2ResponseObservable(parent ResponseObservable, mapper MappingResponse2ResponseFuncFactory) ResponseObservable {
	return &MappingResponse2ResponseObservable{
		parent: parent,
		mapper: mapper,
	}
}

func MapResponse2ResponseObserveDirect(parent ResponseObservable, mapper MappingResponse2ResponseFunc) ResponseObservable {
	return MapResponse2ResponseObservable(parent, func(ResponseObserver) MappingResponse2ResponseFunc {
		return mapper
	})
}

func MapResponse2ResponseObserveNext(parent ResponseObservable, mapper func(*http.Response) *http.Response) ResponseObservable {
	return MapResponse2ResponseObservable(parent, func(ResponseObserver) MappingResponse2ResponseFunc {
		return func(next *http.Response, err error, complete bool, observer ResponseObserver) {
			var mapped *http.Response
			if err == nil && !complete {
				mapped = mapper(next)
			}
			PassthroughResponse(mapped, err, complete, observer)
		}
	},
	)
}

type flatMapResponse2Response struct {
	parent ResponseObservable
	mapper func(*http.Response) ResponseObservable
}

func (f *flatMapResponse2Response) Subscribe(observer ResponseObserver) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe(ResponseObserverFunc(func(next *http.Response, err error, complete bool) {
		switch {
		case err != nil:
			wg.Wait()
			observer.Error(err)
		case complete:
			wg.Wait()
			observer.Complete()
		default:
			wg.Add(1)
			observable := f.mapper(next)
			stream := (&ResponseStream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &ResponseStream{ignoreCompletionFilter().Response(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

// Map maps values in this stream to another value.
func (s *ResponseStream) Map(f func(*http.Response) *http.Response) *ResponseStream {
	return FromResponseObservable(MapResponse2ResponseObserveNext(s, f))
}

func (s *ResponseStream) FlatMap(f func(*http.Response) ResponseObservable) *ResponseStream {
	return &ResponseStream{&flatMapResponse2Response{s, f}}
}

type MappingResponse2StringFunc func(next *http.Response, err error, complete bool, observer StringObserver)
type MappingResponse2StringFuncFactory func(observer StringObserver) MappingResponse2StringFunc

type MappingResponse2StringObservable struct {
	parent ResponseObservable
	mapper MappingResponse2StringFuncFactory
}

func (f *MappingResponse2StringObservable) Subscribe(observer StringObserver) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe(ResponseObserverFunc(func(next *http.Response, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func MapResponse2StringObservable(parent ResponseObservable, mapper MappingResponse2StringFuncFactory) StringObservable {
	return &MappingResponse2StringObservable{
		parent: parent,
		mapper: mapper,
	}
}

func MapResponse2StringObserveDirect(parent ResponseObservable, mapper MappingResponse2StringFunc) StringObservable {
	return MapResponse2StringObservable(parent, func(StringObserver) MappingResponse2StringFunc {
		return mapper
	})
}

func MapResponse2StringObserveNext(parent ResponseObservable, mapper func(*http.Response) string) StringObservable {
	return MapResponse2StringObservable(parent, func(StringObserver) MappingResponse2StringFunc {
		return func(next *http.Response, err error, complete bool, observer StringObserver) {
			var mapped string
			if err == nil && !complete {
				mapped = mapper(next)
			}
			PassthroughString(mapped, err, complete, observer)
		}
	},
	)
}

type flatMapResponse2String struct {
	parent ResponseObservable
	mapper func(*http.Response) StringObservable
}

func (f *flatMapResponse2String) Subscribe(observer StringObserver) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe(ResponseObserverFunc(func(next *http.Response, err error, complete bool) {
		switch {
		case err != nil:
			wg.Wait()
			observer.Error(err)
		case complete:
			wg.Wait()
			observer.Complete()
		default:
			wg.Add(1)
			observable := f.mapper(next)
			stream := (&StringStream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &StringStream{ignoreCompletionFilter().String(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

// MapString maps this stream to an StringStream via f.
func (s *ResponseStream) MapString(f func(*http.Response) string) *StringStream {
	return FromStringObservable(MapResponse2StringObserveNext(s, f))
}

func (s *ResponseStream) FlatMapString(f func(*http.Response) StringObservable) *StringStream {
	return &StringStream{&flatMapResponse2String{s, f}}
}

type MappingResponse2IntFunc func(next *http.Response, err error, complete bool, observer IntObserver)
type MappingResponse2IntFuncFactory func(observer IntObserver) MappingResponse2IntFunc

type MappingResponse2IntObservable struct {
	parent ResponseObservable
	mapper MappingResponse2IntFuncFactory
}

func (f *MappingResponse2IntObservable) Subscribe(observer IntObserver) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe(ResponseObserverFunc(func(next *http.Response, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func MapResponse2IntObservable(parent ResponseObservable, mapper MappingResponse2IntFuncFactory) IntObservable {
	return &MappingResponse2IntObservable{
		parent: parent,
		mapper: mapper,
	}
}

func MapResponse2IntObserveDirect(parent ResponseObservable, mapper MappingResponse2IntFunc) IntObservable {
	return MapResponse2IntObservable(parent, func(IntObserver) MappingResponse2IntFunc {
		return mapper
	})
}

func MapResponse2IntObserveNext(parent ResponseObservable, mapper func(*http.Response) int) IntObservable {
	return MapResponse2IntObservable(parent, func(IntObserver) MappingResponse2IntFunc {
		return func(next *http.Response, err error, complete bool, observer IntObserver) {
			var mapped int
			if err == nil && !complete {
				mapped = mapper(next)
			}
			PassthroughInt(mapped, err, complete, observer)
		}
	},
	)
}

type flatMapResponse2Int struct {
	parent ResponseObservable
	mapper func(*http.Response) IntObservable
}

func (f *flatMapResponse2Int) Subscribe(observer IntObserver) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe(ResponseObserverFunc(func(next *http.Response, err error, complete bool) {
		switch {
		case err != nil:
			wg.Wait()
			observer.Error(err)
		case complete:
			wg.Wait()
			observer.Complete()
		default:
			wg.Add(1)
			observable := f.mapper(next)
			stream := (&IntStream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &IntStream{ignoreCompletionFilter().Int(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

// MapInt maps this stream to an IntStream via f.
func (s *ResponseStream) MapInt(f func(*http.Response) int) *IntStream {
	return FromIntObservable(MapResponse2IntObserveNext(s, f))
}

func (s *ResponseStream) FlatMapInt(f func(*http.Response) IntObservable) *IntStream {
	return &IntStream{&flatMapResponse2Int{s, f}}
}

type StringObserver interface {
	Next(string)
	TerminationObserver
}

// A StringSubscriber represents a subscribed StringObserver.
type StringSubscriber interface {
	Subscription
	StringObserver
}

type implStringSubscriber struct {
	Subscription
	StringObserver
}

func StringObserverAsGenericObserver(observer StringObserver) GenericObserver {
	return NewGenericObserverFunc(func(next interface{}, err error, complete bool) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			observer.Next(next.(string))
		}
	})
}

func GenericObserverAsStringObserver(observer GenericObserver) StringObserver {
	return StringObserverFunc(func(next string, err error, complete bool) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			observer.Next(next)
		}
	})
}

type StringObservableFactory func(observer StringObserver, subscription Subscription)

func (f StringObservableFactory) Subscribe(observer StringObserver) Subscription {
	subscription := NewGenericSubscription()
	go f(observer, subscription)
	return subscription
}

// CreateString calls f(observer, subscription) to produce values for a stream.
func CreateString(f func(observer StringObserver, subscription Subscription)) *StringStream {
	return FromStringObservable(StringObservableFactory(f))
}

// Repeat value count times.
func RepeatString(value string, count int) *StringStream {
	return CreateString(func(observer StringObserver, subscription Subscription) {
		for i := 0; i < count; i++ {
			if subscription.Closed() {
				return
			}
			observer.Next(value)
		}
		observer.Complete()
	})
}

// StartString is designed to be used with functions that return a
// (string, error) tuple.
//
// If the error is non-nil the returned StringStream will be that error,
// otherwise it will be a single-value stream of string.
func StartString(f func() (string, error)) *StringStream {
	return CreateString(func(observer StringObserver, subscription Subscription) {
		if v, err := f(); err != nil {
			observer.Error(err)
		} else {
			observer.Next(v)
			observer.Complete()
		}
	})
}

func PassthroughString(next string, err error, complete bool, observer StringObserver) {
	switch {
	case err != nil:
		observer.Error(err)
	case complete:
		observer.Complete()
	default:
		observer.Next(next)
	}
}

var zeroString = *new(string)

type StringObserverFunc func(string, error, bool)

func (f StringObserverFunc) Next(next string) { f(next, nil, false) }
func (f StringObserverFunc) Error(err error)  { f(zeroString, err, false) }
func (f StringObserverFunc) Complete()        { f(zeroString, nil, true) }

type StringObservable interface {
	Subscribe(StringObserver) Subscription
}

// Convert a GenericObservableFilter to a StringObservable
func (f GenericObservableFilterFactory) String(parent StringObservable) StringObservable {
	return MapString2StringObservable(parent, func(observer StringObserver) MappingString2StringFunc {
		gobserver := StringObserverAsGenericObserver(observer)
		filter := f(gobserver)
		return func(next string, err error, complete bool, observer StringObserver) {
			filter(next, err, complete, gobserver)
		}
	},
	)
}

func NeverString() *StringStream {
	return CreateString(func(observer StringObserver, subscription Subscription) {})
}

func EmptyString() *StringStream {
	return CreateString(func(observer StringObserver, subscription Subscription) {
		observer.Complete()
	})
}

func ThrowString(err error) *StringStream {
	return CreateString(func(observer StringObserver, subscription Subscription) {
		observer.Error(err)
	})
}

func FromStringArray(array []string) *StringStream {
	return CreateString(func(observer StringObserver, subscription Subscription) {
		for _, v := range array {
			if subscription.Closed() {
				return
			}
			observer.Next(v)
		}
		observer.Complete()
		subscription.Close()
	})
}

func FromStrings(array ...string) *StringStream {
	return FromStringArray(array)
}

func JustString(element string) *StringStream {
	return FromStringArray([]string{element})
}

func MergeString(observables ...StringObservable) *StringStream {
	if len(observables) == 0 {
		return EmptyString()
	}
	return (&StringStream{observables[0]}).Merge(observables[1:]...)
}

func MergeStringDelayError(observables ...StringObservable) *StringStream {
	if len(observables) == 0 {
		return EmptyString()
	}
	return (&StringStream{observables[0]}).MergeDelayError(observables[1:]...)
}

func FromStringChannel(ch <-chan string) *StringStream {
	return CreateString(func(observer StringObserver, subscription Subscription) {
		for v := range ch {
			if subscription.Closed() {
				return
			}
			observer.Next(v)
		}
		observer.Complete()
	})
}

type StringStream struct {
	StringObservable
}

func FromStringObservable(observable StringObservable) *StringStream {
	return &StringStream{observable}
}

func (s *StringStream) SubscribeFunc(f func(string, error, bool)) Subscription {
	return s.Subscribe(StringObserverFunc(f))
}

func (s *StringStream) SubscribeNext(f func(v string)) Subscription {
	return s.SubscribeFunc(func(next string, err error, complete bool) {
		if err == nil && !complete {
			f(next)
		}
	})
}

// Distinct removes duplicate elements in the stream.
func (s *StringStream) Distinct() *StringStream {
	return FromStringObservable(distinctFilter().String(s))
}

// ElementAt yields the Nth element of the stream.
func (s *StringStream) ElementAt(n int) *StringStream {
	return FromStringObservable(elementAtFilter(n).String(s))
}

// Filter elements in the stream on a function.
func (s *StringStream) Filter(f func(string) bool) *StringStream {
	return FromStringObservable(filterFilter(func(v interface{}) bool { return f(v.(string)) }).String(s))
}

// Last returns just the first element of the stream.
func (s *StringStream) First() *StringStream {
	return FromStringObservable(firstFilter().String(s))
}

// Last returns just the last element of the stream.
func (s *StringStream) Last() *StringStream {
	return FromStringObservable(lastFilter().String(s))
}

// SkipLast skips the first N elements of the stream.
func (s *StringStream) Skip(n int) *StringStream {
	return FromStringObservable(skipFilter(n).String(s))
}

// SkipLast skips the last N elements of the stream.
func (s *StringStream) SkipLast(n int) *StringStream {
	return FromStringObservable(skipLastFilter(n).String(s))
}

// Take returns just the first N elements of the stream.
func (s *StringStream) Take(n int) *StringStream {
	return FromStringObservable(takeFilter(n).String(s))
}

// TakeLast returns just the last N elements of the stream.
func (s *StringStream) TakeLast(n int) *StringStream {
	return FromStringObservable(takeLastFilter(n).String(s))
}

// IgnoreElements ignores elements of the stream and emits only the completion events.
func (s *StringStream) IgnoreElements() *StringStream {
	return FromStringObservable(ignoreElementsFilter().String(s))
}

func (s *StringStream) Replay(size int, duration time.Duration) *StringStream {
	return FromStringObservable(replayFilter(size, duration).String(s))
}

func (s *StringStream) Sample(duration time.Duration) *StringStream {
	return FromStringObservable(sampleFilter(duration).String(s))
}

func (s *StringStream) Debounce(duration time.Duration) *StringStream {
	return FromStringObservable(debounceFilter(duration).String(s))
}

// Wait for completion of the stream and return any error.
func (s *StringStream) Wait() error {
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next string, err error, complete bool) {
		switch {
		case err != nil:
			errch <- err
		case complete:
			errch <- nil
		default:
		}
	})
	return <-errch
}

func MakeStringSubscriber(observer StringObserver) StringSubscriber {
	if subscriber, ok := observer.(StringSubscriber); ok {
		return subscriber
	}
	return &implStringSubscriber{NewGenericSubscription(), observer}
}

type concatStringSubscriber struct {
	observable  int
	observer    StringObserver
	observables []StringObservable
	Subscription
}

func (c *concatStringSubscriber) Next(next string) {
	c.observer.Next(next)
}

func (c *concatStringSubscriber) Error(err error) {
	c.observer.Error(err)
	c.observable = len(c.observables)
	c.Close()
}

func (c *concatStringSubscriber) Complete() {
	c.observable++
	if c.observable >= len(c.observables) {
		c.observer.Complete()
		c.Close()
		return
	}
	c.observables[c.observable].Subscribe(c)
}

type concatStringObservable struct {
	observables []StringObservable
}

func (m *concatStringObservable) Subscribe(observer StringObserver) Subscription {
	if len(m.observables) == 0 {
		observer.Complete()
		return ClosedSubscription
	}
	subscriber := &concatStringSubscriber{
		observer:     observer,
		Subscription: NewGenericSubscription(),
		observables:  m.observables,
	}
	m.observables[0].Subscribe(subscriber)
	return subscriber
}

func (s *StringStream) Concat(observables ...StringObservable) *StringStream {
	return &StringStream{&concatStringObservable{append([]StringObservable{s}, observables...)}}
}

type mergeStringObservable struct {
	delayError  bool
	observables []StringObservable
}

func (m *mergeStringObservable) Subscribe(observer StringObserver) Subscription {
	subscription := NewGenericSubscription()
	lock := sync.Mutex{}
	completed := 0
	var firstError error
	relay := func(next string, err error, complete bool) {
		lock.Lock()
		defer lock.Unlock()
		if completed >= len(m.observables) {
			return
		}

		switch {
		case err != nil:
			if m.delayError {
				firstError = err
				completed++
			} else {
				observer.Error(err)
				completed = len(m.observables)
			}

		case complete:
			completed++
			if completed == len(m.observables) {
				if firstError != nil {
					observer.Error(firstError)
				} else {
					observer.Complete()
				}
			}
		default:
			observer.Next(next)
		}
	}
	for _, observable := range m.observables {
		observable.Subscribe(StringObserverFunc(relay))
	}
	return subscription
}

// Merge an arbitrary number of observables with this one.
// An error from any of the observables will terminate the merged stream.
func (s *StringStream) Merge(other ...StringObservable) *StringStream {
	if len(other) == 0 {
		return s
	}
	return &StringStream{&mergeStringObservable{false, append(other, s)}}
}

// Merge an arbitrary number of observables with this one.
// Any error will be deferred until all observables terminate.
func (s *StringStream) MergeDelayError(other ...StringObservable) *StringStream {
	if len(other) == 0 {
		return s
	}
	return &StringStream{&mergeStringObservable{true, append(other, s)}}
}

type catchStringObservable struct {
	parent StringObservable
	catch  StringObservable
}

func (r *catchStringObservable) Subscribe(observer StringObserver) Subscription {
	subscription := NewGenericSubscription()
	run := func(next string, err error, complete bool) {
		switch {
		case err != nil:
			r.catch.Subscribe(observer)
		case complete:
			observer.Complete()
		default:
			observer.Next(next)
		}
	}
	r.parent.Subscribe(StringObserverFunc(run))
	return subscription
}

func (s *StringStream) Catch(catch StringObservable) *StringStream {
	return &StringStream{&catchStringObservable{s, catch}}
}

type retryStringObservable struct {
	observable StringObservable
}

type retryStringObserver struct {
	observable StringObservable
	observer   StringObserver
}

func (r *retryStringObserver) retry(next string, err error, complete bool) {
	switch {
	case err != nil:
		r.observable.Subscribe(StringObserverFunc(r.retry))
	case complete:
		r.observer.Complete()
	default:
		r.observer.Next(next)
	}
}

func (r *retryStringObservable) Subscribe(observer StringObserver) Subscription {
	subscription := NewGenericSubscription()
	ro := &retryStringObserver{r.observable, observer}
	r.observable.Subscribe(StringObserverFunc(ro.retry))
	return subscription
}

func (s *StringStream) Retry() *StringStream {
	return &StringStream{&retryStringObservable{s}}
}

// Do applies a function for each value passing through the stream.
func (s *StringStream) Do(f func(next string)) *StringStream {
	return FromStringObservable(MapString2StringObserveNext(s, func(next string) string {
		f(next)
		return next
	}))
}

// DoOnError applies a function for any error on the stream.
func (s *StringStream) DoOnError(f func(err error)) *StringStream {
	return FromStringObservable(MapString2StringObserveDirect(s, func(next string, err error, complete bool, observer StringObserver) {
		if err != nil {
			f(err)
		}
		PassthroughString(next, err, complete, observer)
	}))
}

// DoOnComplete applies a function when the stream completes.
func (s *StringStream) DoOnComplete(f func()) *StringStream {
	return FromStringObservable(MapString2StringObserveDirect(s, func(next string, err error, complete bool, observer StringObserver) {
		if complete {
			f()
		}
		PassthroughString(next, err, complete, observer)
	}))
}

func (s *StringStream) Reduce(initial string, reducer func(string, string) string) *StringStream {
	value := initial
	return FromStringObservable(MapString2StringObserveDirect(s, func(next string, err error, complete bool, observer StringObserver) {
		switch {
		case err != nil:
			observer.Next(value)
			observer.Error(err)
		case complete:
			observer.Next(value)
			observer.Complete()
		default:
			value = reducer(value, next)
		}
	}))
}

func (s *StringStream) Scan(initial string, f func(string, string) string) *StringStream {
	value := initial
	return FromStringObservable(MapString2StringObserveDirect(s, func(next string, err error, complete bool, observer StringObserver) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			value = f(value, next)
			observer.Next(value)
		}
	}))
}

type timeoutString struct {
	parent  StringObservable
	timeout time.Duration
}

func (t *timeoutString) Subscribe(observer StringObserver) Subscription {
	subscription := NewChannelSubscription()
	cancel := t.parent.Subscribe(observer)
	go func() {
		select {
		case <-time.After(t.timeout):
			observer.Error(ErrTimeout)
			_ = cancel.Close()
			_ = subscription.Close()
		case <-subscription:
			_ = cancel.Close()
		}
	}()
	return subscription
}

func (s *StringStream) Timeout(timeout time.Duration) *StringStream {
	return &StringStream{&timeoutString{s, timeout}}

}

// ToOneWithError blocks until the stream emits exactly one value. Otherwise, it errors.
func (s *StringStream) ToOneWithError() (string, error) {
	valuech := make(chan string, 1)
	errch := make(chan error, 1)
	FromStringObservable(oneFilter().String(s)).SubscribeFunc(func(next string, err error, complete bool) {
		if err != nil {
			errch <- err
		} else if !complete {
			valuech <- next
		}
	})
	select {
	case value := <-valuech:
		return value, nil
	case err := <-errch:
		return zeroString, err
	}
}

// ToOne blocks and returns the only value emitted by the stream, or the zero
// value if an error occurs.
func (s *StringStream) ToOne() string {
	value, _ := s.ToOneWithError()
	return value
}

// ToArrayWithError collects all values from the stream into an array,
// returning it and any error.
func (s *StringStream) ToArrayWithError() ([]string, error) {
	array := []string{}
	completech := make(chan bool, 1)
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next string, err error, complete bool) {
		switch {
		case err != nil:
			errch <- err
		case complete:
			completech <- true
		default:
			array = append(array, next)
		}
	})
	select {
	case <-completech:
		return array, nil
	case err := <-errch:
		return array, err
	}
}

// ToArray blocks and returns the values from the stream in an array.
func (s *StringStream) ToArray() []string {
	out, _ := s.ToArrayWithError()
	return out
}

// ToChannelWithError returns value and error channels corresponding to the stream elements and any error.
func (s *StringStream) ToChannelWithError() (<-chan string, <-chan error) {
	ch := make(chan string, 1)
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next string, err error, complete bool) {
		switch {
		case err != nil:
			errch <- err
			close(errch)
			close(ch)
		case complete:
			close(ch)
		default:
			ch <- next
		}
	})
	return ch, errch
}

func (s *StringStream) ToChannel() <-chan string {
	ch, _ := s.ToChannelWithError()
	return ch
}

// Count returns an IntStream with the count of elements in this stream.
func (s *StringStream) Count() *IntStream {
	count := 0
	return FromIntObservable(MapString2IntObserveDirect(s, func(next string, err error, complete bool, observer IntObserver) {
		switch {
		case err != nil:
			observer.Next(count)
			observer.Error(err)
		case complete:
			observer.Next(count)
			observer.Complete()
		default:
			count++
		}
	}))
}

type MappingString2ResponseFunc func(next string, err error, complete bool, observer ResponseObserver)
type MappingString2ResponseFuncFactory func(observer ResponseObserver) MappingString2ResponseFunc

type MappingString2ResponseObservable struct {
	parent StringObservable
	mapper MappingString2ResponseFuncFactory
}

func (f *MappingString2ResponseObservable) Subscribe(observer ResponseObserver) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe(StringObserverFunc(func(next string, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func MapString2ResponseObservable(parent StringObservable, mapper MappingString2ResponseFuncFactory) ResponseObservable {
	return &MappingString2ResponseObservable{
		parent: parent,
		mapper: mapper,
	}
}

func MapString2ResponseObserveDirect(parent StringObservable, mapper MappingString2ResponseFunc) ResponseObservable {
	return MapString2ResponseObservable(parent, func(ResponseObserver) MappingString2ResponseFunc {
		return mapper
	})
}

func MapString2ResponseObserveNext(parent StringObservable, mapper func(string) *http.Response) ResponseObservable {
	return MapString2ResponseObservable(parent, func(ResponseObserver) MappingString2ResponseFunc {
		return func(next string, err error, complete bool, observer ResponseObserver) {
			var mapped *http.Response
			if err == nil && !complete {
				mapped = mapper(next)
			}
			PassthroughResponse(mapped, err, complete, observer)
		}
	},
	)
}

type flatMapString2Response struct {
	parent StringObservable
	mapper func(string) ResponseObservable
}

func (f *flatMapString2Response) Subscribe(observer ResponseObserver) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe(StringObserverFunc(func(next string, err error, complete bool) {
		switch {
		case err != nil:
			wg.Wait()
			observer.Error(err)
		case complete:
			wg.Wait()
			observer.Complete()
		default:
			wg.Add(1)
			observable := f.mapper(next)
			stream := (&ResponseStream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &ResponseStream{ignoreCompletionFilter().Response(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

// MapResponse maps this stream to an ResponseStream via f.
func (s *StringStream) MapResponse(f func(string) *http.Response) *ResponseStream {
	return FromResponseObservable(MapString2ResponseObserveNext(s, f))
}

func (s *StringStream) FlatMapResponse(f func(string) ResponseObservable) *ResponseStream {
	return &ResponseStream{&flatMapString2Response{s, f}}
}

type MappingString2StringFunc func(next string, err error, complete bool, observer StringObserver)
type MappingString2StringFuncFactory func(observer StringObserver) MappingString2StringFunc

type MappingString2StringObservable struct {
	parent StringObservable
	mapper MappingString2StringFuncFactory
}

func (f *MappingString2StringObservable) Subscribe(observer StringObserver) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe(StringObserverFunc(func(next string, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func MapString2StringObservable(parent StringObservable, mapper MappingString2StringFuncFactory) StringObservable {
	return &MappingString2StringObservable{
		parent: parent,
		mapper: mapper,
	}
}

func MapString2StringObserveDirect(parent StringObservable, mapper MappingString2StringFunc) StringObservable {
	return MapString2StringObservable(parent, func(StringObserver) MappingString2StringFunc {
		return mapper
	})
}

func MapString2StringObserveNext(parent StringObservable, mapper func(string) string) StringObservable {
	return MapString2StringObservable(parent, func(StringObserver) MappingString2StringFunc {
		return func(next string, err error, complete bool, observer StringObserver) {
			var mapped string
			if err == nil && !complete {
				mapped = mapper(next)
			}
			PassthroughString(mapped, err, complete, observer)
		}
	},
	)
}

type flatMapString2String struct {
	parent StringObservable
	mapper func(string) StringObservable
}

func (f *flatMapString2String) Subscribe(observer StringObserver) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe(StringObserverFunc(func(next string, err error, complete bool) {
		switch {
		case err != nil:
			wg.Wait()
			observer.Error(err)
		case complete:
			wg.Wait()
			observer.Complete()
		default:
			wg.Add(1)
			observable := f.mapper(next)
			stream := (&StringStream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &StringStream{ignoreCompletionFilter().String(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

// Map maps values in this stream to another value.
func (s *StringStream) Map(f func(string) string) *StringStream {
	return FromStringObservable(MapString2StringObserveNext(s, f))
}

func (s *StringStream) FlatMap(f func(string) StringObservable) *StringStream {
	return &StringStream{&flatMapString2String{s, f}}
}

type MappingString2IntFunc func(next string, err error, complete bool, observer IntObserver)
type MappingString2IntFuncFactory func(observer IntObserver) MappingString2IntFunc

type MappingString2IntObservable struct {
	parent StringObservable
	mapper MappingString2IntFuncFactory
}

func (f *MappingString2IntObservable) Subscribe(observer IntObserver) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe(StringObserverFunc(func(next string, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func MapString2IntObservable(parent StringObservable, mapper MappingString2IntFuncFactory) IntObservable {
	return &MappingString2IntObservable{
		parent: parent,
		mapper: mapper,
	}
}

func MapString2IntObserveDirect(parent StringObservable, mapper MappingString2IntFunc) IntObservable {
	return MapString2IntObservable(parent, func(IntObserver) MappingString2IntFunc {
		return mapper
	})
}

func MapString2IntObserveNext(parent StringObservable, mapper func(string) int) IntObservable {
	return MapString2IntObservable(parent, func(IntObserver) MappingString2IntFunc {
		return func(next string, err error, complete bool, observer IntObserver) {
			var mapped int
			if err == nil && !complete {
				mapped = mapper(next)
			}
			PassthroughInt(mapped, err, complete, observer)
		}
	},
	)
}

type flatMapString2Int struct {
	parent StringObservable
	mapper func(string) IntObservable
}

func (f *flatMapString2Int) Subscribe(observer IntObserver) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe(StringObserverFunc(func(next string, err error, complete bool) {
		switch {
		case err != nil:
			wg.Wait()
			observer.Error(err)
		case complete:
			wg.Wait()
			observer.Complete()
		default:
			wg.Add(1)
			observable := f.mapper(next)
			stream := (&IntStream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &IntStream{ignoreCompletionFilter().Int(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

// MapInt maps this stream to an IntStream via f.
func (s *StringStream) MapInt(f func(string) int) *IntStream {
	return FromIntObservable(MapString2IntObserveNext(s, f))
}

func (s *StringStream) FlatMapInt(f func(string) IntObservable) *IntStream {
	return &IntStream{&flatMapString2Int{s, f}}
}

type IntObserver interface {
	Next(int)
	TerminationObserver
}

// A IntSubscriber represents a subscribed IntObserver.
type IntSubscriber interface {
	Subscription
	IntObserver
}

type implIntSubscriber struct {
	Subscription
	IntObserver
}

func IntObserverAsGenericObserver(observer IntObserver) GenericObserver {
	return NewGenericObserverFunc(func(next interface{}, err error, complete bool) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			observer.Next(next.(int))
		}
	})
}

func GenericObserverAsIntObserver(observer GenericObserver) IntObserver {
	return IntObserverFunc(func(next int, err error, complete bool) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			observer.Next(next)
		}
	})
}

type IntObservableFactory func(observer IntObserver, subscription Subscription)

func (f IntObservableFactory) Subscribe(observer IntObserver) Subscription {
	subscription := NewGenericSubscription()
	go f(observer, subscription)
	return subscription
}

// CreateInt calls f(observer, subscription) to produce values for a stream.
func CreateInt(f func(observer IntObserver, subscription Subscription)) *IntStream {
	return FromIntObservable(IntObservableFactory(f))
}

// Repeat value count times.
func RepeatInt(value int, count int) *IntStream {
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		for i := 0; i < count; i++ {
			if subscription.Closed() {
				return
			}
			observer.Next(value)
		}
		observer.Complete()
	})
}

// StartInt is designed to be used with functions that return a
// (int, error) tuple.
//
// If the error is non-nil the returned IntStream will be that error,
// otherwise it will be a single-value stream of int.
func StartInt(f func() (int, error)) *IntStream {
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		if v, err := f(); err != nil {
			observer.Error(err)
		} else {
			observer.Next(v)
			observer.Complete()
		}
	})
}

func PassthroughInt(next int, err error, complete bool, observer IntObserver) {
	switch {
	case err != nil:
		observer.Error(err)
	case complete:
		observer.Complete()
	default:
		observer.Next(next)
	}
}

var zeroInt = *new(int)

type IntObserverFunc func(int, error, bool)

func (f IntObserverFunc) Next(next int)   { f(next, nil, false) }
func (f IntObserverFunc) Error(err error) { f(zeroInt, err, false) }
func (f IntObserverFunc) Complete()       { f(zeroInt, nil, true) }

type IntObservable interface {
	Subscribe(IntObserver) Subscription
}

// Convert a GenericObservableFilter to a IntObservable
func (f GenericObservableFilterFactory) Int(parent IntObservable) IntObservable {
	return MapInt2IntObservable(parent, func(observer IntObserver) MappingInt2IntFunc {
		gobserver := IntObserverAsGenericObserver(observer)
		filter := f(gobserver)
		return func(next int, err error, complete bool, observer IntObserver) {
			filter(next, err, complete, gobserver)
		}
	},
	)
}

func NeverInt() *IntStream {
	return CreateInt(func(observer IntObserver, subscription Subscription) {})
}

func EmptyInt() *IntStream {
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		observer.Complete()
	})
}

func ThrowInt(err error) *IntStream {
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		observer.Error(err)
	})
}

func FromIntArray(array []int) *IntStream {
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		for _, v := range array {
			if subscription.Closed() {
				return
			}
			observer.Next(v)
		}
		observer.Complete()
		subscription.Close()
	})
}

func FromInts(array ...int) *IntStream {
	return FromIntArray(array)
}

func JustInt(element int) *IntStream {
	return FromIntArray([]int{element})
}

func MergeInt(observables ...IntObservable) *IntStream {
	if len(observables) == 0 {
		return EmptyInt()
	}
	return (&IntStream{observables[0]}).Merge(observables[1:]...)
}

func MergeIntDelayError(observables ...IntObservable) *IntStream {
	if len(observables) == 0 {
		return EmptyInt()
	}
	return (&IntStream{observables[0]}).MergeDelayError(observables[1:]...)
}

func FromIntChannel(ch <-chan int) *IntStream {
	return CreateInt(func(observer IntObserver, subscription Subscription) {
		for v := range ch {
			if subscription.Closed() {
				return
			}
			observer.Next(v)
		}
		observer.Complete()
	})
}

type IntStream struct {
	IntObservable
}

func FromIntObservable(observable IntObservable) *IntStream {
	return &IntStream{observable}
}

func (s *IntStream) SubscribeFunc(f func(int, error, bool)) Subscription {
	return s.Subscribe(IntObserverFunc(f))
}

func (s *IntStream) SubscribeNext(f func(v int)) Subscription {
	return s.SubscribeFunc(func(next int, err error, complete bool) {
		if err == nil && !complete {
			f(next)
		}
	})
}

// Distinct removes duplicate elements in the stream.
func (s *IntStream) Distinct() *IntStream {
	return FromIntObservable(distinctFilter().Int(s))
}

// ElementAt yields the Nth element of the stream.
func (s *IntStream) ElementAt(n int) *IntStream {
	return FromIntObservable(elementAtFilter(n).Int(s))
}

// Filter elements in the stream on a function.
func (s *IntStream) Filter(f func(int) bool) *IntStream {
	return FromIntObservable(filterFilter(func(v interface{}) bool { return f(v.(int)) }).Int(s))
}

// Last returns just the first element of the stream.
func (s *IntStream) First() *IntStream {
	return FromIntObservable(firstFilter().Int(s))
}

// Last returns just the last element of the stream.
func (s *IntStream) Last() *IntStream {
	return FromIntObservable(lastFilter().Int(s))
}

// SkipLast skips the first N elements of the stream.
func (s *IntStream) Skip(n int) *IntStream {
	return FromIntObservable(skipFilter(n).Int(s))
}

// SkipLast skips the last N elements of the stream.
func (s *IntStream) SkipLast(n int) *IntStream {
	return FromIntObservable(skipLastFilter(n).Int(s))
}

// Take returns just the first N elements of the stream.
func (s *IntStream) Take(n int) *IntStream {
	return FromIntObservable(takeFilter(n).Int(s))
}

// TakeLast returns just the last N elements of the stream.
func (s *IntStream) TakeLast(n int) *IntStream {
	return FromIntObservable(takeLastFilter(n).Int(s))
}

// IgnoreElements ignores elements of the stream and emits only the completion events.
func (s *IntStream) IgnoreElements() *IntStream {
	return FromIntObservable(ignoreElementsFilter().Int(s))
}

func (s *IntStream) Replay(size int, duration time.Duration) *IntStream {
	return FromIntObservable(replayFilter(size, duration).Int(s))
}

func (s *IntStream) Sample(duration time.Duration) *IntStream {
	return FromIntObservable(sampleFilter(duration).Int(s))
}

func (s *IntStream) Debounce(duration time.Duration) *IntStream {
	return FromIntObservable(debounceFilter(duration).Int(s))
}

// Wait for completion of the stream and return any error.
func (s *IntStream) Wait() error {
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next int, err error, complete bool) {
		switch {
		case err != nil:
			errch <- err
		case complete:
			errch <- nil
		default:
		}
	})
	return <-errch
}

func MakeIntSubscriber(observer IntObserver) IntSubscriber {
	if subscriber, ok := observer.(IntSubscriber); ok {
		return subscriber
	}
	return &implIntSubscriber{NewGenericSubscription(), observer}
}

type concatIntSubscriber struct {
	observable  int
	observer    IntObserver
	observables []IntObservable
	Subscription
}

func (c *concatIntSubscriber) Next(next int) {
	c.observer.Next(next)
}

func (c *concatIntSubscriber) Error(err error) {
	c.observer.Error(err)
	c.observable = len(c.observables)
	c.Close()
}

func (c *concatIntSubscriber) Complete() {
	c.observable++
	if c.observable >= len(c.observables) {
		c.observer.Complete()
		c.Close()
		return
	}
	c.observables[c.observable].Subscribe(c)
}

type concatIntObservable struct {
	observables []IntObservable
}

func (m *concatIntObservable) Subscribe(observer IntObserver) Subscription {
	if len(m.observables) == 0 {
		observer.Complete()
		return ClosedSubscription
	}
	subscriber := &concatIntSubscriber{
		observer:     observer,
		Subscription: NewGenericSubscription(),
		observables:  m.observables,
	}
	m.observables[0].Subscribe(subscriber)
	return subscriber
}

func (s *IntStream) Concat(observables ...IntObservable) *IntStream {
	return &IntStream{&concatIntObservable{append([]IntObservable{s}, observables...)}}
}

type mergeIntObservable struct {
	delayError  bool
	observables []IntObservable
}

func (m *mergeIntObservable) Subscribe(observer IntObserver) Subscription {
	subscription := NewGenericSubscription()
	lock := sync.Mutex{}
	completed := 0
	var firstError error
	relay := func(next int, err error, complete bool) {
		lock.Lock()
		defer lock.Unlock()
		if completed >= len(m.observables) {
			return
		}

		switch {
		case err != nil:
			if m.delayError {
				firstError = err
				completed++
			} else {
				observer.Error(err)
				completed = len(m.observables)
			}

		case complete:
			completed++
			if completed == len(m.observables) {
				if firstError != nil {
					observer.Error(firstError)
				} else {
					observer.Complete()
				}
			}
		default:
			observer.Next(next)
		}
	}
	for _, observable := range m.observables {
		observable.Subscribe(IntObserverFunc(relay))
	}
	return subscription
}

// Merge an arbitrary number of observables with this one.
// An error from any of the observables will terminate the merged stream.
func (s *IntStream) Merge(other ...IntObservable) *IntStream {
	if len(other) == 0 {
		return s
	}
	return &IntStream{&mergeIntObservable{false, append(other, s)}}
}

// Merge an arbitrary number of observables with this one.
// Any error will be deferred until all observables terminate.
func (s *IntStream) MergeDelayError(other ...IntObservable) *IntStream {
	if len(other) == 0 {
		return s
	}
	return &IntStream{&mergeIntObservable{true, append(other, s)}}
}

type catchIntObservable struct {
	parent IntObservable
	catch  IntObservable
}

func (r *catchIntObservable) Subscribe(observer IntObserver) Subscription {
	subscription := NewGenericSubscription()
	run := func(next int, err error, complete bool) {
		switch {
		case err != nil:
			r.catch.Subscribe(observer)
		case complete:
			observer.Complete()
		default:
			observer.Next(next)
		}
	}
	r.parent.Subscribe(IntObserverFunc(run))
	return subscription
}

func (s *IntStream) Catch(catch IntObservable) *IntStream {
	return &IntStream{&catchIntObservable{s, catch}}
}

type retryIntObservable struct {
	observable IntObservable
}

type retryIntObserver struct {
	observable IntObservable
	observer   IntObserver
}

func (r *retryIntObserver) retry(next int, err error, complete bool) {
	switch {
	case err != nil:
		r.observable.Subscribe(IntObserverFunc(r.retry))
	case complete:
		r.observer.Complete()
	default:
		r.observer.Next(next)
	}
}

func (r *retryIntObservable) Subscribe(observer IntObserver) Subscription {
	subscription := NewGenericSubscription()
	ro := &retryIntObserver{r.observable, observer}
	r.observable.Subscribe(IntObserverFunc(ro.retry))
	return subscription
}

func (s *IntStream) Retry() *IntStream {
	return &IntStream{&retryIntObservable{s}}
}

// Do applies a function for each value passing through the stream.
func (s *IntStream) Do(f func(next int)) *IntStream {
	return FromIntObservable(MapInt2IntObserveNext(s, func(next int) int {
		f(next)
		return next
	}))
}

// DoOnError applies a function for any error on the stream.
func (s *IntStream) DoOnError(f func(err error)) *IntStream {
	return FromIntObservable(MapInt2IntObserveDirect(s, func(next int, err error, complete bool, observer IntObserver) {
		if err != nil {
			f(err)
		}
		PassthroughInt(next, err, complete, observer)
	}))
}

// DoOnComplete applies a function when the stream completes.
func (s *IntStream) DoOnComplete(f func()) *IntStream {
	return FromIntObservable(MapInt2IntObserveDirect(s, func(next int, err error, complete bool, observer IntObserver) {
		if complete {
			f()
		}
		PassthroughInt(next, err, complete, observer)
	}))
}

func (s *IntStream) Reduce(initial int, reducer func(int, int) int) *IntStream {
	value := initial
	return FromIntObservable(MapInt2IntObserveDirect(s, func(next int, err error, complete bool, observer IntObserver) {
		switch {
		case err != nil:
			observer.Next(value)
			observer.Error(err)
		case complete:
			observer.Next(value)
			observer.Complete()
		default:
			value = reducer(value, next)
		}
	}))
}

func (s *IntStream) Scan(initial int, f func(int, int) int) *IntStream {
	value := initial
	return FromIntObservable(MapInt2IntObserveDirect(s, func(next int, err error, complete bool, observer IntObserver) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			value = f(value, next)
			observer.Next(value)
		}
	}))
}

type timeoutInt struct {
	parent  IntObservable
	timeout time.Duration
}

func (t *timeoutInt) Subscribe(observer IntObserver) Subscription {
	subscription := NewChannelSubscription()
	cancel := t.parent.Subscribe(observer)
	go func() {
		select {
		case <-time.After(t.timeout):
			observer.Error(ErrTimeout)
			_ = cancel.Close()
			_ = subscription.Close()
		case <-subscription:
			_ = cancel.Close()
		}
	}()
	return subscription
}

func (s *IntStream) Timeout(timeout time.Duration) *IntStream {
	return &IntStream{&timeoutInt{s, timeout}}

}

// ToOneWithError blocks until the stream emits exactly one value. Otherwise, it errors.
func (s *IntStream) ToOneWithError() (int, error) {
	valuech := make(chan int, 1)
	errch := make(chan error, 1)
	FromIntObservable(oneFilter().Int(s)).SubscribeFunc(func(next int, err error, complete bool) {
		if err != nil {
			errch <- err
		} else if !complete {
			valuech <- next
		}
	})
	select {
	case value := <-valuech:
		return value, nil
	case err := <-errch:
		return zeroInt, err
	}
}

// ToOne blocks and returns the only value emitted by the stream, or the zero
// value if an error occurs.
func (s *IntStream) ToOne() int {
	value, _ := s.ToOneWithError()
	return value
}

// ToArrayWithError collects all values from the stream into an array,
// returning it and any error.
func (s *IntStream) ToArrayWithError() ([]int, error) {
	array := []int{}
	completech := make(chan bool, 1)
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next int, err error, complete bool) {
		switch {
		case err != nil:
			errch <- err
		case complete:
			completech <- true
		default:
			array = append(array, next)
		}
	})
	select {
	case <-completech:
		return array, nil
	case err := <-errch:
		return array, err
	}
}

// ToArray blocks and returns the values from the stream in an array.
func (s *IntStream) ToArray() []int {
	out, _ := s.ToArrayWithError()
	return out
}

// ToChannelWithError returns value and error channels corresponding to the stream elements and any error.
func (s *IntStream) ToChannelWithError() (<-chan int, <-chan error) {
	ch := make(chan int, 1)
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next int, err error, complete bool) {
		switch {
		case err != nil:
			errch <- err
			close(errch)
			close(ch)
		case complete:
			close(ch)
		default:
			ch <- next
		}
	})
	return ch, errch
}

func (s *IntStream) ToChannel() <-chan int {
	ch, _ := s.ToChannelWithError()
	return ch
}

// Count returns an IntStream with the count of elements in this stream.
func (s *IntStream) Count() *IntStream {
	count := 0
	return FromIntObservable(MapInt2IntObserveDirect(s, func(next int, err error, complete bool, observer IntObserver) {
		switch {
		case err != nil:
			observer.Next(count)
			observer.Error(err)
		case complete:
			observer.Next(count)
			observer.Complete()
		default:
			count++
		}
	}))
}

func (s *IntStream) Average() *IntStream {
	var sum int
	var count int
	return FromIntObservable(MapInt2IntObserveDirect(s, func(next int, err error, complete bool, observer IntObserver) {
		switch {
		case err != nil:
			observer.Next(sum / count)
			observer.Error(err)
		case complete:
			observer.Next(sum / count)
			observer.Complete()
		default:
			sum += next
			count++
		}
	}))
}

func (s *IntStream) Sum() *IntStream {
	var sum int
	return FromIntObservable(MapInt2IntObserveDirect(s, func(next int, err error, complete bool, observer IntObserver) {
		switch {
		case err != nil:
			observer.Next(sum)
			observer.Error(err)
		case complete:
			observer.Next(sum)
			observer.Complete()
		default:
			sum += next
		}
	}))
}

func (s *IntStream) Min() *IntStream {
	started := false
	var min int
	return FromIntObservable(MapInt2IntObserveDirect(s, func(next int, err error, complete bool, observer IntObserver) {
		switch {
		case err != nil:
			if started {
				observer.Next(min)
			}
			observer.Error(err)
		case complete:
			if started {
				observer.Next(min)
			}
			observer.Complete()
		default:
			if started {
				if min > next {
					min = next
				}
			} else {
				min = next
				started = true
			}
		}
	}))
}

func (s *IntStream) Max() *IntStream {
	started := false
	var max int
	return FromIntObservable(MapInt2IntObserveDirect(s, func(next int, err error, complete bool, observer IntObserver) {
		switch {
		case err != nil:
			if started {
				observer.Next(max)
			}
			observer.Error(err)
		case complete:
			if started {
				observer.Next(max)
			}
			observer.Complete()
		default:
			if started {
				if max <= next {
					max = next
				}
			} else {
				max = next
				started = true
			}
		}
	}))
}

type MappingInt2ResponseFunc func(next int, err error, complete bool, observer ResponseObserver)
type MappingInt2ResponseFuncFactory func(observer ResponseObserver) MappingInt2ResponseFunc

type MappingInt2ResponseObservable struct {
	parent IntObservable
	mapper MappingInt2ResponseFuncFactory
}

func (f *MappingInt2ResponseObservable) Subscribe(observer ResponseObserver) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe(IntObserverFunc(func(next int, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func MapInt2ResponseObservable(parent IntObservable, mapper MappingInt2ResponseFuncFactory) ResponseObservable {
	return &MappingInt2ResponseObservable{
		parent: parent,
		mapper: mapper,
	}
}

func MapInt2ResponseObserveDirect(parent IntObservable, mapper MappingInt2ResponseFunc) ResponseObservable {
	return MapInt2ResponseObservable(parent, func(ResponseObserver) MappingInt2ResponseFunc {
		return mapper
	})
}

func MapInt2ResponseObserveNext(parent IntObservable, mapper func(int) *http.Response) ResponseObservable {
	return MapInt2ResponseObservable(parent, func(ResponseObserver) MappingInt2ResponseFunc {
		return func(next int, err error, complete bool, observer ResponseObserver) {
			var mapped *http.Response
			if err == nil && !complete {
				mapped = mapper(next)
			}
			PassthroughResponse(mapped, err, complete, observer)
		}
	},
	)
}

type flatMapInt2Response struct {
	parent IntObservable
	mapper func(int) ResponseObservable
}

func (f *flatMapInt2Response) Subscribe(observer ResponseObserver) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe(IntObserverFunc(func(next int, err error, complete bool) {
		switch {
		case err != nil:
			wg.Wait()
			observer.Error(err)
		case complete:
			wg.Wait()
			observer.Complete()
		default:
			wg.Add(1)
			observable := f.mapper(next)
			stream := (&ResponseStream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &ResponseStream{ignoreCompletionFilter().Response(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

// MapResponse maps this stream to an ResponseStream via f.
func (s *IntStream) MapResponse(f func(int) *http.Response) *ResponseStream {
	return FromResponseObservable(MapInt2ResponseObserveNext(s, f))
}

func (s *IntStream) FlatMapResponse(f func(int) ResponseObservable) *ResponseStream {
	return &ResponseStream{&flatMapInt2Response{s, f}}
}

type MappingInt2StringFunc func(next int, err error, complete bool, observer StringObserver)
type MappingInt2StringFuncFactory func(observer StringObserver) MappingInt2StringFunc

type MappingInt2StringObservable struct {
	parent IntObservable
	mapper MappingInt2StringFuncFactory
}

func (f *MappingInt2StringObservable) Subscribe(observer StringObserver) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe(IntObserverFunc(func(next int, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func MapInt2StringObservable(parent IntObservable, mapper MappingInt2StringFuncFactory) StringObservable {
	return &MappingInt2StringObservable{
		parent: parent,
		mapper: mapper,
	}
}

func MapInt2StringObserveDirect(parent IntObservable, mapper MappingInt2StringFunc) StringObservable {
	return MapInt2StringObservable(parent, func(StringObserver) MappingInt2StringFunc {
		return mapper
	})
}

func MapInt2StringObserveNext(parent IntObservable, mapper func(int) string) StringObservable {
	return MapInt2StringObservable(parent, func(StringObserver) MappingInt2StringFunc {
		return func(next int, err error, complete bool, observer StringObserver) {
			var mapped string
			if err == nil && !complete {
				mapped = mapper(next)
			}
			PassthroughString(mapped, err, complete, observer)
		}
	},
	)
}

type flatMapInt2String struct {
	parent IntObservable
	mapper func(int) StringObservable
}

func (f *flatMapInt2String) Subscribe(observer StringObserver) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe(IntObserverFunc(func(next int, err error, complete bool) {
		switch {
		case err != nil:
			wg.Wait()
			observer.Error(err)
		case complete:
			wg.Wait()
			observer.Complete()
		default:
			wg.Add(1)
			observable := f.mapper(next)
			stream := (&StringStream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &StringStream{ignoreCompletionFilter().String(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

// MapString maps this stream to an StringStream via f.
func (s *IntStream) MapString(f func(int) string) *StringStream {
	return FromStringObservable(MapInt2StringObserveNext(s, f))
}

func (s *IntStream) FlatMapString(f func(int) StringObservable) *StringStream {
	return &StringStream{&flatMapInt2String{s, f}}
}

type MappingInt2IntFunc func(next int, err error, complete bool, observer IntObserver)
type MappingInt2IntFuncFactory func(observer IntObserver) MappingInt2IntFunc

type MappingInt2IntObservable struct {
	parent IntObservable
	mapper MappingInt2IntFuncFactory
}

func (f *MappingInt2IntObservable) Subscribe(observer IntObserver) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe(IntObserverFunc(func(next int, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func MapInt2IntObservable(parent IntObservable, mapper MappingInt2IntFuncFactory) IntObservable {
	return &MappingInt2IntObservable{
		parent: parent,
		mapper: mapper,
	}
}

func MapInt2IntObserveDirect(parent IntObservable, mapper MappingInt2IntFunc) IntObservable {
	return MapInt2IntObservable(parent, func(IntObserver) MappingInt2IntFunc {
		return mapper
	})
}

func MapInt2IntObserveNext(parent IntObservable, mapper func(int) int) IntObservable {
	return MapInt2IntObservable(parent, func(IntObserver) MappingInt2IntFunc {
		return func(next int, err error, complete bool, observer IntObserver) {
			var mapped int
			if err == nil && !complete {
				mapped = mapper(next)
			}
			PassthroughInt(mapped, err, complete, observer)
		}
	},
	)
}

type flatMapInt2Int struct {
	parent IntObservable
	mapper func(int) IntObservable
}

func (f *flatMapInt2Int) Subscribe(observer IntObserver) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe(IntObserverFunc(func(next int, err error, complete bool) {
		switch {
		case err != nil:
			wg.Wait()
			observer.Error(err)
		case complete:
			wg.Wait()
			observer.Complete()
		default:
			wg.Add(1)
			observable := f.mapper(next)
			stream := (&IntStream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &IntStream{ignoreCompletionFilter().Int(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

// Map maps values in this stream to another value.
func (s *IntStream) Map(f func(int) int) *IntStream {
	return FromIntObservable(MapInt2IntObserveNext(s, f))
}

func (s *IntStream) FlatMap(f func(int) IntObservable) *IntStream {
	return &IntStream{&flatMapInt2Int{s, f}}
}
