func (s *ResponseStream) Distinct() *ResponseStream {
	return FromResponseObservable(distinctFilter().Response(s))
}
Distinct removes duplicate elements in the stream.
package rx
import (
	"errors"
	"io"
	"sync"
	"sync/atomic"
	"time"
	"net/http"
)
var ErrTimeout = errors.New("timeout")
var MaxReplaySize = 16384
type Subscription interface {
	io.Closer
	Closed() bool
}
type closedSubscription struct{}
func (closedSubscription) Close() error { return nil }
func (closedSubscription) Closed() bool { return true }
var ClosedSubscription Subscription = closedSubscription{}
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
}
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
	}
}
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
}
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)
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
}
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
}
func CreateResponse(f func(observer ResponseObserver, subscription Subscription)) *ResponseStream {
	return FromResponseObservable(ResponseObservableFactory(f))
}
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()
	})
}
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
}
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)
		}
	})
}