package main

import (
	"fmt"
	"go/ast"
	"go/parser"
	"io/ioutil"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"strings"

	"github.com/alecthomas/template"
	"gopkg.in/alecthomas/kingpin.v2"
)

var (
	commonTypes = strings.Fields("int string float32 float64")
	baseTypes   = strings.Fields("bool rune byte string uint int uint8 int8 uint16 int16 uint32 int32 uint64 int64 float32 float64 complex64 complex128 time.Time time.Duration []byte")
)

var (
	packageArg      = kingpin.Arg("package", "Go package.").Required().String()
	typesArg        = kingpin.Arg("types", "List of types to provide Reactive types for.").Strings()
	commonTypesFlag = kingpin.Flag("common-types", "Define extensions for common Go types.").Action(func(*kingpin.ParseContext) error {
		*typesArg = append(*typesArg, "int", "string", "float32", "float64")
		return nil
	}).Bool()
	baseTypesFlag = kingpin.Flag("base-types", "Define extensions for all base Go types.").Action(func(*kingpin.ParseContext) error {
		*typesArg = append(*typesArg, baseTypes...)
		return nil
	}).Bool()
	importsFlag   = kingpin.Flag("import", "Extra imports.").PlaceHolder("PKG...").Strings()
	outputFlag    = kingpin.Flag("output", "File to write to.").Short('o').String()
	debugFlag     = kingpin.Flag("debug", "Debug mode.").Bool()
	maxReplayFlag = kingpin.Flag("max-replay", "Maximum size of replayed data.").Default("16384").Int()
)

var reactTemplate = `// Package {{.Package}} implements ReactiveX extensions for Go.
package {{.Package}}

//go:generate {{.Generate|Join}}

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

import (
	"errors"
	"time"
	"sync"
	"sync/atomic"
{{range .Imports}}
	"{{.}}"
{{end}}\
)

// 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 = {{.MaxReplaySize}}

// A Subscription to an observable.
type Subscription interface {
	// Dispose unsubscribes from the subscription.
	Dispose()
	// Disposed returns true if this subscription has been unsubscribed.
	Disposed() bool
}

// SubscriptionEvents provides lifecycle event callbacks for a Subscription.
type SubscriptionEvents interface {
	OnUnsubscribe(func ())
}

// A Subscription that is already closed.
type closedSubscription struct {}
func (closedSubscription) Dispose() {}
func (closedSubscription) Disposed() bool { return true }

// ClosedSubscription always returns true for Disposed()
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.Dispose()
	}
}

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

func (l *LinkedSubscription) Disposed() bool {
	l.lock.Lock()
	defer l.lock.Unlock()
	if l.linked != nil {
		return l.linked.Disposed()
	}
	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) Dispose() {
	defer recover()
	close(c)
}

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

func (c ChannelSubscription) OnUnsubscribe(handler func ()) {
	go func() {
		<-c
		handler()
	}()
}

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

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

func (t *GenericSubscription) Dispose() {
	atomic.StoreInt32((*int32)(t), 1)
}

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

type CallbackSubscription func()

func (c *CallbackSubscription) Dispose() {
	if *c != nil {
		(*c)()
		*c = nil
	}
}

func (c *CallbackSubscription) Disposed() bool {
	return *c == nil
}


// 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.Disposed() {
				return
			}
			observer.Next(i)
		}
		observer.Complete()
		subscription.Dispose()
	})
}

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

{{range $type := .Types}}
{{with $name := .|TypeName}}
type {{$name}}Observer interface {
	Next({{$type}})
	TerminationObserver
}

// A {{$name}}Subscriber represents a subscribed {{$name}}Observer.
type {{$name}}Subscriber interface {
	Subscription
	{{$name}}Observer
}

type impl{{$name}}Subscriber struct {
	Subscription
	{{$name}}Observer
}

func {{$name}}ObserverAsGenericObserver(observer {{$name}}Observer) 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.({{$type}}))
		}
	})
}

func GenericObserverAs{{$name}}Observer(observer GenericObserver) {{$name}}Observer {
	return {{$name}}ObserverFunc(func(next {{$type}}, err error, complete bool) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			observer.Next(next)
		}
	})
}

type {{$name}}ObservableFactory func (observer {{$name}}Observer, subscription Subscription)

func (f {{$name}}ObservableFactory) Subscribe(observer {{$name}}Observer) Subscription {
	subscription := NewGenericSubscription()
	go f(observer, subscription)
	return subscription
}

// Create{{$name}} calls f(observer, subscription) to produce values for a stream.
func Create{{$name}}(f func (observer {{$name}}Observer, subscription Subscription)) *{{$name}}Stream {
	return From{{$name}}Observable({{$name}}ObservableFactory(f))
}

// Repeat value count times.
func Repeat{{$name}}(value {{$type}}, count int) *{{$name}}Stream {
	return Create{{$name}}(func (observer {{$name}}Observer, subscription Subscription) {
		for i := 0; i < count; i++ {
			if subscription.Disposed() {
				return
			}
			observer.Next(value)
		}
		observer.Complete()
	})
}

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

func Passthrough{{$name}}(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
	switch {
	case err != nil:
		observer.Error(err)
	case complete:
		observer.Complete()
	default:
		observer.Next(next)
	}
}

var zero{{$name}} = *new({{$type}})

type {{$name}}ObserverFunc func({{$type}}, error, bool)

func (f {{$name}}ObserverFunc) Next(next {{$type}}) { f(next, nil, false) }
func (f {{$name}}ObserverFunc) Error(err error)  { f(zero{{$name}}, err, false) }
func (f {{$name}}ObserverFunc) Complete()        { f(zero{{$name}}, nil, true) }

type {{$name}}Observable interface {
	Subscribe({{$name}}Observer) Subscription
}

// Convert a GenericObservableFilter to a {{$name}}Observable
func (f GenericObservableFilterFactory) {{$name}}(parent {{$name}}Observable) {{$name}}Observable {
	return Map{{$name}}2{{$name}}Observable(parent, func(observer {{$name}}Observer) Mapping{{$name}}2{{$name}}Func {
			gobserver := {{$name}}ObserverAsGenericObserver(observer)
			filter := f(gobserver)
			return func(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
				filter(next, err, complete, gobserver)
			}
		},
	)
}

func Never{{$name}}() *{{$name}}Stream {
	return Create{{$name}}(func (observer {{$name}}Observer, subscription Subscription) {})
}

func Empty{{$name}}() *{{$name}}Stream {
	return Create{{$name}}(func (observer {{$name}}Observer, subscription Subscription) {
		observer.Complete()
	})
}

func Throw{{$name}}(err error) *{{$name}}Stream {
	return Create{{$name}}(func (observer {{$name}}Observer, subscription Subscription) {
		observer.Error(err)
	})
}

func From{{$name}}Array(array []{{$type}}) *{{$name}}Stream {
	return Create{{$name}}(func (observer {{$name}}Observer, subscription Subscription) {
		for _, v := range array {
			if subscription.Disposed() {
				return
			}
			observer.Next(v)
		}
		observer.Complete()
		subscription.Dispose()
	})
}

func From{{$name}}s(array ...{{$type}}) *{{$name}}Stream {
	return From{{$name}}Array(array)
}

func Just{{$name}}(element {{$type}}) *{{$name}}Stream {
	return From{{$name}}Array([]{{$type}}{element})
}

func Merge{{$name}}(observables ... {{$name}}Observable) *{{$name}}Stream {
	if len(observables) == 0 {
		return Empty{{$name}}()
	}
	return (&{{$name}}Stream{observables[0]}).Merge(observables[1:]...)
}

func Merge{{$name}}DelayError(observables ... {{$name}}Observable) *{{$name}}Stream {
	if len(observables) == 0 {
		return Empty{{$name}}()
	}
	return (&{{$name}}Stream{observables[0]}).MergeDelayError(observables[1:]...)
}

func From{{$name}}Channel(ch <-chan {{$type}}) *{{$name}}Stream {
	return Create{{$name}}(func (observer {{$name}}Observer, subscription Subscription) {
		for v := range ch {
			if subscription.Disposed() {
				return
			}
			observer.Next(v)
		}
		observer.Complete()
	})
}

type {{$name}}Stream struct {
	{{$name}}Observable
}

func From{{$name}}Observable(observable {{$name}}Observable) *{{$name}}Stream {
	return &{{$name}}Stream{observable}
}

func (s *{{$name}}Stream) SubscribeFunc(f func({{$type}}, error, bool)) Subscription {
	return s.Subscribe({{$name}}ObserverFunc(f))
}

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

// Distinct removes duplicate elements in the stream.
func (s *{{$name}}Stream) Distinct() *{{$name}}Stream {
	return From{{$name}}Observable(distinctFilter().{{$name}}(s))
}

// ElementAt yields the Nth element of the stream.
func (s *{{$name}}Stream) ElementAt(n int) *{{$name}}Stream {
	return From{{$name}}Observable(elementAtFilter(n).{{$name}}(s))
}

// Filter elements in the stream on a function.
func (s *{{$name}}Stream) Filter(f func({{$type}}) bool) *{{$name}}Stream {
	return From{{$name}}Observable(filterFilter(func(v interface{}) bool { return f(v.({{$type}})) }).{{$name}}(s))
}

// Last returns just the first element of the stream.
func (s *{{$name}}Stream) First() *{{$name}}Stream {
	return From{{$name}}Observable(firstFilter().{{$name}}(s))
}

// Last returns just the last element of the stream.
func (s *{{$name}}Stream) Last() *{{$name}}Stream {
	return From{{$name}}Observable(lastFilter().{{$name}}(s))
}

// SkipLast skips the first N elements of the stream.
func (s *{{$name}}Stream) Skip(n int) *{{$name}}Stream {
	return From{{$name}}Observable(skipFilter(n).{{$name}}(s))
}

// SkipLast skips the last N elements of the stream.
func (s *{{$name}}Stream) SkipLast(n int) *{{$name}}Stream {
	return From{{$name}}Observable(skipLastFilter(n).{{$name}}(s))
}

// Take returns just the first N elements of the stream.
func (s *{{$name}}Stream) Take(n int) *{{$name}}Stream {
	return From{{$name}}Observable(takeFilter(n).{{$name}}(s))
}

// TakeLast returns just the last N elements of the stream.
func (s *{{$name}}Stream) TakeLast(n int) *{{$name}}Stream {
	return From{{$name}}Observable(takeLastFilter(n).{{$name}}(s))
}

// IgnoreElements ignores elements of the stream and emits only the completion events.
func (s *{{$name}}Stream) IgnoreElements() *{{$name}}Stream {
	return From{{$name}}Observable(ignoreElementsFilter().{{$name}}(s))
}

func (s *{{$name}}Stream) Replay(size int, duration time.Duration) *{{$name}}Stream {
	return From{{$name}}Observable(replayFilter(size, duration).{{$name}}(s))
}

func (s *{{$name}}Stream) Sample(duration time.Duration) *{{$name}}Stream {
	return From{{$name}}Observable(sampleFilter(duration).{{$name}}(s))
}

func (s *{{$name}}Stream) Debounce(duration time.Duration) *{{$name}}Stream {
	return From{{$name}}Observable(debounceFilter(duration).{{$name}}(s))
}

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

func Make{{$name}}Subscriber(observer {{$name}}Observer) {{$name}}Subscriber {
	if subscriber, ok := observer.({{$name}}Subscriber); ok {
		return subscriber
	}
	return &impl{{$name}}Subscriber{NewGenericSubscription(), observer}
}

type concat{{$name}}Subscriber struct {
	observable   int
	observer     {{$name}}Observer
	observables  []{{$name}}Observable
	Subscription
}

func (c *concat{{$name}}Subscriber) Next(next {{$type}}) {
	c.observer.Next(next)
}

func (c *concat{{$name}}Subscriber) Error(err error) {
	c.observer.Error(err)
	c.observable = len(c.observables)
	c.Dispose()
}

func (c *concat{{$name}}Subscriber) Complete() {
	c.observable++
	if c.observable >= len(c.observables) {
		c.observer.Complete()
		c.Dispose()
		return
	}
	c.observables[c.observable].Subscribe(c)
}

type concat{{$name}}Observable struct {
	observables []{{$name}}Observable
}

func (m *concat{{$name}}Observable) Subscribe(observer {{$name}}Observer) Subscription {
	if len(m.observables) == 0 {
		observer.Complete()
		return ClosedSubscription
	}
	subscriber := &concat{{$name}}Subscriber{
		observer:     observer,
		Subscription: NewGenericSubscription(),
		observables:  m.observables,
	}
	m.observables[0].Subscribe(subscriber)
	return subscriber
}

func (s *{{$name}}Stream) Concat(observables ... {{$name}}Observable) *{{$name}}Stream {
	return &{{$name}}Stream{&concat{{$name}}Observable{append([]{{$name}}Observable{s}, observables...)} }
}

type merge{{$name}}Observable struct {
	delayError bool
	observables []{{$name}}Observable
}

func (m *merge{{$name}}Observable) Subscribe(observer {{$name}}Observer) Subscription {
	subscription := NewGenericSubscription()
	lock := sync.Mutex{}
	completed := 0
	var firstError error
	relay := func(next {{$type}}, 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({{$name}}ObserverFunc(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 *{{$name}}Stream) Merge(other ... {{$name}}Observable) *{{$name}}Stream {
	if len(other) == 0 {
		return s
	}
	return &{{$name}}Stream{&merge{{$name}}Observable{false, append(other, s) } }
}

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

type catch{{$name}}Observable struct {
	parent {{$name}}Observable
	catch {{$name}}Observable
}

func (r *catch{{$name}}Observable) Subscribe(observer {{$name}}Observer) Subscription {
	subscription := NewGenericSubscription()
	run := func(next {{$type}}, err error, complete bool) {
		switch {
		case err != nil:
			r.catch.Subscribe(observer)
		case complete:
			observer.Complete()
		default:
			observer.Next(next)
		}
	}
	r.parent.Subscribe({{$name}}ObserverFunc(run))
	return subscription
}

func (s *{{$name}}Stream) Catch(catch {{$name}}Observable) *{{$name}}Stream {
	return &{{$name}}Stream{ &catch{{$name}}Observable{s, catch} }
}

type retry{{$name}}Observable struct {
	observable {{$name}}Observable
}

type retry{{$name}}Observer struct {
	observable {{$name}}Observable
	observer {{$name}}Observer
}

func (r *retry{{$name}}Observer) retry(next {{$type}}, err error, complete bool) {
	switch {
	case err != nil:
		r.observable.Subscribe({{$name}}ObserverFunc(r.retry))
	case complete:
		r.observer.Complete()
	default:
		r.observer.Next(next)
	}
}

func (r *retry{{$name}}Observable) Subscribe(observer {{$name}}Observer) Subscription {
	subscription := NewGenericSubscription()
	ro := &retry{{$name}}Observer{r.observable, observer}
	r.observable.Subscribe({{$name}}ObserverFunc(ro.retry))
	return subscription
}

func (s *{{$name}}Stream) Retry() *{{$name}}Stream {
	return &{{$name}}Stream{ &retry{{$name}}Observable{s} }
}

// Do applies a function for each value passing through the stream.
func (s *{{$name}}Stream) Do(f func(next {{$type}})) *{{$name}}Stream {
	return From{{$name}}Observable(Map{{$name}}2{{$name}}ObserveNext(s, func(next {{$type}}) {{$type}} {
		f(next)
		return next
	}))
}

// DoOnError applies a function for any error on the stream.
func (s *{{$name}}Stream) DoOnError(f func(err error)) *{{$name}}Stream {
	return From{{$name}}Observable(Map{{$name}}2{{$name}}ObserveDirect(s, func(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
		if err != nil {
			f(err)
		}
		Passthrough{{$name}}(next, err, complete, observer)
	}))
}

// DoOnComplete applies a function when the stream completes.
func (s *{{$name}}Stream) DoOnComplete(f func()) *{{$name}}Stream {
	return From{{$name}}Observable(Map{{$name}}2{{$name}}ObserveDirect(s, func(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
		if complete {
			f()
		}
		Passthrough{{$name}}(next, err, complete, observer)
	}))
}

func (s *{{$name}}Stream) Reduce(initial {{$type}}, reducer func ({{$type}}, {{$type}}) {{$type}}) *{{$name}}Stream {
	value := initial
	return From{{$name}}Observable(Map{{$name}}2{{$name}}ObserveDirect(s, func(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
		switch {
		case err != nil:
			observer.Next(value)
			observer.Error(err)
		case complete:
			observer.Next(value)
			observer.Complete()
		default:
			value = reducer(value, next)
		}
	}))
}

func (s *{{$name}}Stream) Scan(initial {{$type}}, f func ({{$type}}, {{$type}}) {{$type}}) *{{$name}}Stream {
	value := initial
	return From{{$name}}Observable(Map{{$name}}2{{$name}}ObserveDirect(s, func(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
		switch {
		case err != nil:
			observer.Error(err)
		case complete:
			observer.Complete()
		default:
			value = f(value, next)
			observer.Next(value)
		}
	}))
}

type timeout{{$name}} struct {
	parent {{$name}}Observable
	timeout time.Duration
}

func (t *timeout{{$name}}) Subscribe(observer {{$name}}Observer) Subscription {
	subscription := NewChannelSubscription()
	cancel := t.parent.Subscribe(observer)
	go func() {
		select {
		case <-time.After(t.timeout):
			observer.Error(ErrTimeout)
			cancel.Dispose()
			subscription.Dispose()
		case <-subscription:
			cancel.Dispose()
		}
	}()
	return subscription
}

func (s *{{$name}}Stream) Timeout(timeout time.Duration) *{{$name}}Stream {
	return &{{$name}}Stream{&timeout{{$name}}{s, timeout}}
}

type forked{{$name}}Stream struct {
	lock sync.Mutex
	parent {{$name}}Observable
	observers []{{$name}}Observer
}

func (f *forked{{$name}}Stream) Subscribe(observer {{$name}}Observer) Subscription {
	f.lock.Lock()
	defer f.lock.Unlock()
	i := len(f.observers)
	f.observers = append(f.observers, observer)
	sub := new(CallbackSubscription)
	*sub = CallbackSubscription(func() {
		f.lock.Lock()
		defer f.lock.Unlock()
		f.observers[i] = nil
	})
	return sub
}

// Fork replicates each event from the parent to every subscriber of the fork.
func (s *{{$name}}Stream) Fork() *{{$name}}Stream {
	f := &forked{{$name}}Stream{parent: s}
	go s.Subscribe({{$name}}ObserverFunc(func(n {{$type}}, err error, complete bool) {
		f.lock.Lock()
		defer f.lock.Unlock()
		for _, o := range f.observers {
			if o == nil {
				continue
			}
			switch {
			case err != nil:
				o.Error(err)
			case complete:
				o.Complete()
			default:
				o.Next(n)
			}
		}
	}))
	return &{{$name}}Stream{f}
}

// ToOneWithError blocks until the stream emits exactly one value. Otherwise, it errors.
func (s *{{$name}}Stream) ToOneWithError() ({{$type}}, error) {
	valuech := make(chan {{$type}}, 1)
	errch := make(chan error, 1)
	From{{$name}}Observable(oneFilter().{{$name}}(s)).SubscribeFunc(func (next {{$type}}, 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 zero{{$name}}, err
	}
}

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

// ToArrayWithError collects all values from the stream into an array,
// returning it and any error.
func (s *{{$name}}Stream) ToArrayWithError() ([]{{$type}}, error) {
	array := []{{$type}}{}
	completech := make(chan bool, 1)
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next {{$type}}, 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 *{{$name}}Stream) ToArray() []{{$type}} {
	out, _ := s.ToArrayWithError()
	return out
}

// ToChannelWithError returns value and error channels corresponding to the stream elements and any error.
func (s *{{$name}}Stream) ToChannelWithError() (<-chan {{$type}}, <-chan error) {
	ch := make(chan {{$type}}, 1)
	errch := make(chan error, 1)
	s.SubscribeFunc(func(next {{$type}}, 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 *{{$name}}Stream) ToChannel() <-chan {{$type}} {
	ch, _ := s.ToChannelWithError()
	return ch
}

// Count returns an IntStream with the count of elements in this stream.
func (s *{{$name}}Stream) Count() *IntStream {
	count := 0
	return FromIntObservable(Map{{$name}}2IntObserveDirect(s, func(next {{$type}}, 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++
		}
	}))
}

{{if $type|IsNumeric}}\
func (s *{{$name}}Stream) Average() *{{$name}}Stream {
	var sum {{$type}}
	var count {{$type}}
	return From{{$name}}Observable(Map{{$name}}2{{$name}}ObserveDirect(s, func(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
		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 *{{$name}}Stream) Sum() *{{$name}}Stream {
	var sum {{$type}}
	return From{{$name}}Observable(Map{{$name}}2{{$name}}ObserveDirect(s, func(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
		switch {
		case err != nil:
			observer.Next(sum)
			observer.Error(err)
		case complete:
			observer.Next(sum)
			observer.Complete()
		default:
			sum += next
		}
	}))
}

func (s *{{$name}}Stream) Min() *{{$name}}Stream {
	started := false
	var min {{$type}}
	return From{{$name}}Observable(Map{{$name}}2{{$name}}ObserveDirect(s, func(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
		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 *{{$name}}Stream) Max() *{{$name}}Stream {
	started := false
	var max {{$type}}
	return From{{$name}}Observable(Map{{$name}}2{{$name}}ObserveDirect(s, func(next {{$type}}, err error, complete bool, observer {{$name}}Observer) {
		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
			}
		}
	}))
}
{{end}}\

{{range $other := $.Types}}\
{{with $otherName := .|TypeName}}\
{{with $mapName := printf "%s2%s" $name $otherName}}
type Mapping{{$mapName}}Func func(next {{$type}}, err error, complete bool, observer {{$otherName}}Observer)
type Mapping{{$mapName}}FuncFactory func (observer {{$otherName}}Observer) Mapping{{$mapName}}Func

type Mapping{{$mapName}}Observable struct {
	parent  {{$name}}Observable
	mapper Mapping{{$mapName}}FuncFactory
}

func (f *Mapping{{$mapName}}Observable) Subscribe(observer {{$otherName}}Observer) Subscription {
	mapper := f.mapper(observer)
	return f.parent.Subscribe({{$name}}ObserverFunc(func(next {{$type}}, err error, complete bool) {
		mapper(next, err, complete, observer)
	}))
}

func Map{{$mapName}}Observable(parent {{$name}}Observable, mapper Mapping{{$mapName}}FuncFactory) {{$otherName}}Observable {
	return &Mapping{{$mapName}}Observable{
		parent:  parent,
		mapper: mapper,
	}
}

func Map{{$mapName}}ObserveDirect(parent {{$name}}Observable, mapper Mapping{{$mapName}}Func) {{$otherName}}Observable {
	return Map{{$mapName}}Observable(parent, func({{$otherName}}Observer) Mapping{{$mapName}}Func {
		return mapper
	})
}

func Map{{$mapName}}ObserveNext(parent {{$name}}Observable, mapper func({{$type}}) {{$other}}) {{$otherName}}Observable {
	return Map{{$mapName}}Observable(parent, func({{$otherName}}Observer) Mapping{{$mapName}}Func {
			return func(next {{$type}}, err error, complete bool, observer {{$otherName}}Observer) {
				var mapped {{$other}}
				if err == nil && !complete {
					mapped = mapper(next)
				}
				Passthrough{{$otherName}}(mapped, err, complete, observer)
			}
		},
	)
}

type flatMap{{$mapName}} struct {
	parent {{$name}}Observable
	mapper func ({{$type}}) {{$otherName}}Observable
}

func (f *flatMap{{$mapName}}) Subscribe(observer {{$otherName}}Observer) Subscription {
	subscription := NewGenericSubscription()
	wg := sync.WaitGroup{}
	f.parent.Subscribe({{$name}}ObserverFunc(func (next {{$type}}, 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 := (&{{$otherName}}Stream{observable}).
				DoOnComplete(func() { wg.Done() }).
				DoOnError(func(error) { wg.Done() })
			stream = &{{$otherName}}Stream{ignoreCompletionFilter().{{$otherName}}(stream)}
			stream.Subscribe(observer)
		}
	}))
	return subscription
}

{{if ne $type $other}}
// Map{{$otherName}} maps this stream to an {{$otherName}}Stream via f.
func (s *{{$name}}Stream) Map{{$otherName}}(f func ({{$type}}) {{$other}}) *{{$otherName}}Stream {
	return From{{$otherName}}Observable(Map{{$mapName}}ObserveNext(s, f))
}

func (s *{{$name}}Stream) FlatMap{{$otherName}}(f func ({{$type}}) {{$otherName}}Observable) *{{$otherName}}Stream {
	return &{{$otherName}}Stream{&flatMap{{$mapName}}{s, f}}
}
{{else}}\
// Map maps values in this stream to another value.
func (s *{{$name}}Stream) Map(f func ({{$type}}) {{$other}}) *{{$name}}Stream {
	return From{{$name}}Observable(Map{{$mapName}}ObserveNext(s, f))
}

func (s *{{$name}}Stream) FlatMap(f func ({{$type}}) {{$name}}Observable) *{{$name}}Stream {
	return &{{$name}}Stream{&flatMap{{$mapName}}{s, f}}
}
{{end}}
{{end}}
{{end}}\
{{end}}
{{end}}\
{{end}}\
`

type Context struct {
	Package       string
	Generate      []string
	Types         []string
	Imports       []string
	MaxReplaySize int
}

func typeName(t ast.Expr) []string {
	switch n := t.(type) {
	case *ast.StarExpr:
		return typeName(n.X)
	case *ast.SelectorExpr:
		// return append(typeName(n.X), typeName(n.Sel)...)
		return typeName(n.Sel)
	case *ast.MapType:
		keys := typeName(n.Key)
		keys = append(keys, typeName(n.Value)...)
		keys = append(keys, "Map")
		return keys
	case *ast.ArrayType:
		return append(typeName(n.Elt), "Slice")
	case *ast.Ident:
		return []string{strings.Title(n.Name)}
	default:
		panic(fmt.Sprintf("unknown expression node %s %s\n", reflect.TypeOf(t), t))
	}
}

func main() {
	kingpin.Parse()
	t := template.Must(template.New("react").Funcs(template.FuncMap{
		"TypeName": func(s string) string {
			st, err := parser.ParseExpr(s)
			kingpin.FatalIfError(err, "invalid type %q", s)
			return strings.Join(typeName(st), "")
		},
		"IsNumeric": func(v string) bool {
			switch v {
			case "byte", "uint", "int", "uint8", "int8", "uint16", "int16", "uint32",
				"int32", "uint64", "int64", "float32", "float64":
				return true
			}
			return false
		},
		"Join": func(a []string) string { return strings.Join(a, " ") },
	}).Parse(reactTemplate))
	// We always include int. As an aside, this kind of verbosity is my
	// biggest peeve with Go. 10 lines compared to basically any other
	// language:
	//
	// 		if *typesArg.Contains("int") { ... }
	//
	// Super annoying.
	found := false
	for _, s := range *typesArg {
		if s == "int" {
			found = true
			break
		}
	}
	if !found {
		*typesArg = append(*typesArg, "int")
	}
	cmd := exec.Command("goimports")
	if *debugFlag {
		cmd = exec.Command("cat")
	}
	outputFile := os.Stdout
	if *outputFlag != "" {
		dir := filepath.Dir(*outputFlag)
		var err error
		outputFile, err = ioutil.TempFile(dir, ".gorx-")
		kingpin.FatalIfError(err, "")
		defer os.Remove(outputFile.Name())
	}
	cmd.Stdout = outputFile
	stdin, err := cmd.StdinPipe()
	kingpin.FatalIfError(err, "")
	kingpin.FatalIfError(cmd.Start(), "")
	err = t.Execute(stdin, &Context{
		Generate:      append([]string{"gorx"}, os.Args[1:]...),
		Package:       *packageArg,
		Types:         *typesArg,
		Imports:       *importsFlag,
		MaxReplaySize: *maxReplayFlag,
	})
	kingpin.FatalIfError(err, "")
	stdin.Close()
	kingpin.FatalIfError(cmd.Wait(), "")
	outputFile.Close()
	if *outputFlag != "" {
		kingpin.FatalIfError(os.Rename(outputFile.Name(), *outputFlag), "")
	}
}
