<|start_filename|>flux/processor_test.go<|end_filename|>
package flux_test

import (
	"context"
	"fmt"
	"io"
	"testing"
	"time"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/flux"
	"github.com/jjeffcaii/reactor-go/hooks"
	"github.com/stretchr/testify/assert"
)

func TestUnicastProcessor_Close(t *testing.T) {
	droppedError := make(chan error)
	droppedNext := make(chan Any)
	hooks.OnErrorDrop(func(e error) {
		droppedError <- e
	})
	hooks.OnNextDrop(func(v reactor.Any) {
		droppedNext <- v
	})
	p := flux.NewUnicastProcessor()

	go func() {
		p.Error(fakeErr)
	}()

	_ = p.(io.Closer).Close()
	select {
	case e := <-droppedError:
		assert.Equal(t, fakeErr, e, "should be fake error")
	case <-droppedNext:
		assert.Fail(t, "unreachable")
	}
	p = flux.NewUnicastProcessor()

	fakeElem := "fake element"
	go func() {
		p.Next(fakeElem)
		p.Complete()
	}()

	_ = p.(io.Closer).Close()
	select {
	case <-droppedError:
		assert.Fail(t, "unreachable")
	case next := <-droppedNext:
		assert.Equal(t, fakeElem, next)
	}
}

func TestUnicastProcessor(t *testing.T) {
	hooks.OnNextDrop(func(v Any) {
		fmt.Println("dropped:", v)
	})
	p := flux.NewUnicastProcessor()

	time.AfterFunc(10*time.Millisecond, func() {
		p.Next(1)
	})

	time.AfterFunc(15*time.Millisecond, func() {
		p.Next(2)
	})
	time.AfterFunc(20*time.Millisecond, func() {
		p.Next(3)
		p.Complete()
	})

	results := make(chan int)

	var su reactor.Subscription
	p.
		DoOnNext(func(v Any) error {
			time.Sleep(30 * time.Millisecond)
			fmt.Println("onNext:", v)
			results <- v.(int)
			su.Request(1)
			return nil
		}).
		DoOnRequest(func(n int) {
			fmt.Println("request:", n)
		}).
		DoOnComplete(func() {
			fmt.Println("complete")
			close(results)
		}).
		Subscribe(context.Background(), reactor.OnSubscribe(func(ctx context.Context, s reactor.Subscription) {
			su = s
			su.Request(1)
		}))

	done := make(chan struct{})
	go func() {
		defer close(done)
		var sum int
		for v := range results {
			sum += v
		}
		assert.Equal(t, 6, sum, "bad sum result")
	}()
	<-done
}


<|start_filename|>mono/switch_if_error_test.go<|end_filename|>
package mono_test

import (
	"context"
	"errors"
	"fmt"
	"testing"
	"time"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/mono"
	"github.com/stretchr/testify/assert"
)

func TestSwitchIfError(t *testing.T) {
	fakeErr := errors.New("fake error")

	// should call switch to another Mono if error occurs
	value, err := mono.Error(fakeErr).
		Map(func(any reactor.Any) (reactor.Any, error) {
			assert.FailNow(t, "should be unreachable")
			return nil, nil
		}).
		SwitchIfError(func(err error) mono.Mono {
			assert.Equal(t, fakeErr, err)
			return mono.Just(1)
		}).
		Map(func(any reactor.Any) (reactor.Any, error) {
			return any.(int) * 2, nil
		}).
		Block(context.Background())
	assert.NoError(t, err)
	assert.Equal(t, 2, value)

	// should not call SwitchIfError when Mono ends with a successful value
	value, err = mono.Just(1).
		Map(func(any reactor.Any) (reactor.Any, error) {
			return any.(int) + 1, nil
		}).
		SwitchIfError(func(err error) mono.Mono {
			assert.FailNow(t, "should be unreachable")
			return mono.Error(err)
		}).
		Map(func(any reactor.Any) (reactor.Any, error) {
			return any.(int) * 2, nil
		}).
		Block(context.Background())
	assert.NoError(t, err)
	assert.Equal(t, 4, value)

	// nothing happened if empty
	value, err = mono.Empty().
		SwitchIfError(func(err error) mono.Mono {
			assert.FailNow(t, "should be unreachable")
			return mono.Error(err)
		}).
		Map(func(any reactor.Any) (reactor.Any, error) {
			assert.FailNow(t, "should be unreachable")
			return nil, nil
		}).
		Block(context.Background())
	assert.NoError(t, err)
	assert.Nil(t, value)

	// check oneshot api
	value, err = mono.ErrorOneshot(fakeErr).
		SwitchIfError(func(err error) mono.Mono {
			assert.Equal(t, fakeErr, err)
			return mono.Just(1)
		}).
		Block(context.Background())
	assert.NoError(t, err)
	assert.Equal(t, 1, value)

	delay := 500 * time.Millisecond

	done := make(chan struct{})
	var end time.Time
	mono.Error(fakeErr).
		SwitchIfError(func(err error) mono.Mono {
			// simulate async mono
			return mono.Delay(delay).
				Map(func(any reactor.Any) (reactor.Any, error) {
					return 1, nil
				}).
				DoFinally(func(s reactor.SignalType) {
					close(done)
				})
		}).
		DoOnNext(func(v reactor.Any) error {
			assert.Equal(t, 1, v)
			return nil
		}).
		DoOnError(func(e error) {
			assert.FailNow(t, "should be unreachable")
		}).
		DoOnComplete(func() {
			end = time.Now()
		}).
		Subscribe(context.Background())
	start := time.Now()
	fmt.Println("should not block here, done will be finished after", delay)
	<-done
	assert.True(t, int64(end.Sub(start)) > int64(delay))
}


<|start_filename|>scheduler/parallel_test.go<|end_filename|>
package scheduler_test

import (
	"fmt"
	"testing"

	"github.com/jjeffcaii/reactor-go/scheduler"
	"github.com/stretchr/testify/assert"
)

func TestParallel(t *testing.T) {
	fmt.Println(scheduler.Parallel().Name())
	defer scheduler.Parallel().Close()
	done := make(chan int)
	_ = scheduler.Parallel().Worker().Do(func() {
		done <- 1
	})
	assert.Equal(t, 1, <-done)
}


<|start_filename|>mono/processor_test.go<|end_filename|>
package mono_test

import (
	"context"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/mono"
	"github.com/jjeffcaii/reactor-go/scheduler"
	"github.com/stretchr/testify/assert"
)

func TestProcessor(t *testing.T) {
	p, s, d := mono.NewProcessor(nil, nil)
	defer d.Dispose()

	go func() {
		now := time.Now()
		s.Success("hello world")
		t.Log("cost:", time.Since(now))
	}()

	const total = 8

	var wg sync.WaitGroup

	wg.Add(total)

	onNext := reactor.OnNext(func(v reactor.Any) error {
		time.Sleep(100 * time.Millisecond)
		t.Log("next:", v)
		return nil
	})

	onComplete := reactor.OnComplete(func() {
		wg.Done()
	})

	for range [total]struct{}{} {
		p.SubscribeOn(scheduler.Parallel()).Subscribe(context.Background(), onNext, onComplete)
	}

	wg.Wait()

}
func TestProcessor_Map(t *testing.T) {
	p, s, d := mono.NewProcessor(nil, nil)
	defer d.Dispose()

	time.AfterFunc(100*time.Millisecond, func() {
		s.Success(1024)
	})

	transform := func(any reactor.Any) (reactor.Any, error) {
		return any.(int) * 2, nil
	}
	v, err := p.Map(transform).Block(context.Background())
	assert.NoError(t, err)
	assert.Equal(t, 2048, v)
}

func TestProcessorNG_Order(t *testing.T) {
	valueChan := make(chan reactor.Any)
	onNext := reactor.OnNext(func(v reactor.Any) error {
		valueChan <- v
		return nil
	})

	p, s, d := mono.NewProcessor(nil, nil)
	defer d.Dispose()
	p.SubscribeOn(scheduler.Parallel()).Subscribe(context.Background(), onNext)

	select {
	case <-valueChan:
		assert.Fail(t, "unreachable")
	default:
	}

	s.Success(123)
	assert.Equal(t, 123, <-valueChan)
}

func TestProcessorSubscriberNG_Cancel(t *testing.T) {
	done := make(chan struct{})
	p, _, d := mono.NewProcessor(nil, nil)
	defer d.Dispose()
	onSubscribe := reactor.OnSubscribe(func(ctx context.Context, su reactor.Subscription) {
		su.Cancel()
	})
	doFinally := func(s reactor.SignalType) {
		close(done)
	}
	onNext := reactor.OnNext(func(v reactor.Any) error {
		assert.Fail(t, "unreachable")
		return nil
	})
	p.DoFinally(doFinally).Subscribe(context.Background(), onSubscribe, onNext)
}

func TestProcessor_Finally(t *testing.T) {
	const total = 2

	var wg sync.WaitGroup
	wg.Add(total)

	cnt := new(int32)
	fakeValue := 1

	for range [total]struct{}{} {
		p, s, _ := mono.NewProcessor(scheduler.Parallel(), func(signalType reactor.SignalType, disposable reactor.Disposable) {
			disposable.Dispose()
			wg.Done()
		})

		go func() {
			s.Success(fakeValue)
		}()

		p.Subscribe(context.Background(), reactor.OnNext(func(v reactor.Any) error {
			assert.Equal(t, fakeValue, v)
			return nil
		}), reactor.OnError(func(e error) {
			assert.NoError(t, e, "should not return error")
		}), reactor.OnComplete(func() {
			atomic.AddInt32(cnt, 1)
		}))
	}
	wg.Wait()

	assert.Equal(t, int32(total), atomic.LoadInt32(cnt))
}


<|start_filename|>hc/option.go<|end_filename|>
package hc

import (
	"encoding/json"
	"errors"
	"io/ioutil"
	"net/http"
)

type BodyParser func(*http.Response) (interface{}, error)

// WithStringResponse decodes the response body as string.
func WithStringResponse() Option {
	return WithResponseBodyParser(func(res *http.Response) (interface{}, error) {
		defer func() {
			if res.Body != nil {
				_ = res.Body.Close()
			}
		}()

		if !isSuccessful(res.StatusCode) {
			b, err := ioutil.ReadAll(res.Body)
			if err == nil {
				err = errors.New(string(b))
			}
			return nil, err
		}

		b, err := ioutil.ReadAll(res.Body)
		if err != nil {
			return nil, err
		}
		return string(b), nil
	})
}

// WithJSONResponse decodes the response body as json.
func WithJSONResponse(gen func() (ptr interface{})) Option {
	return func(o *option) {
		o.bodyParser = func(res *http.Response) (interface{}, error) {
			defer func() {
				if res.Body != nil {
					_ = res.Body.Close()
				}
			}()

			if !isSuccessful(res.StatusCode) {
				b, err := ioutil.ReadAll(res.Body)
				if err != nil {
					return nil, err
				}
				return nil, errors.New(string(b))
			}

			value := gen()

			if err := json.NewDecoder(res.Body).Decode(value); err != nil {
				return nil, err
			}
			return value, nil
		}
	}
}

// WithRequestHijack can be used to customize the http request.
func WithRequestHijack(h func(*http.Request)) Option {
	return func(o *option) {
		o.reqChain = append(o.reqChain, h)
	}
}

// WithResponseBodyParser set the response body parser.
func WithResponseBodyParser(bp BodyParser) Option {
	return func(o *option) {
		o.bodyParser = bp
	}
}

type option struct {
	bodyParser BodyParser
	reqChain   []func(*http.Request)
}

func isSuccessful(code int) bool {
	return code > 199 && code < 300
}


<|start_filename|>hooks/hooks_test.go<|end_filename|>
package hooks_test

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/hooks"
	"github.com/jjeffcaii/reactor-go/mono"
	"github.com/stretchr/testify/assert"
)

type Any = reactor.Any

func TestCreate_OnNextDrop(t *testing.T) {
	var droppedValue int
	var droppedError error
	hooks.OnNextDrop(func(i Any) {
		droppedValue = i.(int)
	})
	hooks.OnErrorDrop(func(e error) {
		droppedError = e
	})

	done := make(chan struct{})

	mono.
		Create(func(context context.Context, sink mono.Sink) {
			sink.Success(1)
			// Should be dropped below
			sink.Success(2)
			sink.Error(errors.New("mock"))
			close(done)
		}).
		DoOnNext(func(v Any) error {
			assert.Equal(t, 1, v.(int), "bad doOnNext")
			return nil
		}).
		Subscribe(context.Background())
	<-done
	assert.Equal(t, 2, droppedValue, "bad onNextDrop")
	assert.Error(t, droppedError, "no OnErrorDrop happened")
}

func TestProcessor_OnNextDrop(t *testing.T) {
	done := make(chan struct{})
	var hasDrop bool
	hooks.OnNextDrop(func(i Any) {
		defer close(done)
		assert.Equal(t, 2, i.(int), "bad onNextDrop")
		hasDrop = true
	})
	p, s, d := mono.NewProcessor(nil, nil)
	defer d.Dispose()
	time.AfterFunc(100*time.Millisecond, func() {
		s.Success(1)
	})
	time.AfterFunc(200*time.Millisecond, func() {
		s.Success(2)
	})
	p.
		DoOnNext(func(v Any) error {
			assert.Equal(t, 1, v.(int), "bad doOnNext")
			return nil
		}).
		Subscribe(context.Background())
	<-done
	assert.True(t, hasDrop, "no doNextDrop happened")
}


<|start_filename|>flux/op_switch_on_first_test.go<|end_filename|>
package flux_test

import (
	"context"
	"fmt"
	"testing"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/flux"
	"github.com/stretchr/testify/assert"
)

func TestSwitchOnFirst(t *testing.T) {
	// greater than first element and multiply 2
	origin := []Any{3, 8, 7, 6, 5, 4, 3, 2, 1}

	var expected []int
	for _, it := range origin {
		if it.(int) > origin[0].(int) {
			expected = append(expected, it.(int)*2)
		}
	}
	var actual []int
	flux.Just(origin...).
		SwitchOnFirst(func(s flux.Signal, f flux.Flux) flux.Flux {
			if first, ok := s.Value(); ok {
				return f.
					Filter(func(i Any) bool {
						return i.(int) > first.(int)
					}).
					Map(func(i Any) (Any, error) {
						return i.(int) * 2, nil
					})
			}
			return f
		}).
		DoOnNext(func(v Any) error {
			actual = append(actual, v.(int))
			fmt.Println("next:", v)
			return nil
		}).
		DoFinally(func(s reactor.SignalType) {
			fmt.Println("finally:", s)
		}).
		Subscribe(context.Background())
	assert.Equal(t, expected, actual, "bad result")
}


<|start_filename|>reactor_test.go<|end_filename|>
package reactor_test

import (
	"testing"

	"github.com/jjeffcaii/reactor-go"
	"github.com/stretchr/testify/assert"
)

func TestSignalType_String(t *testing.T) {
	for _, v := range []struct {
		Signal reactor.SignalType
		String string
	}{
		{reactor.SignalTypeCancel, "CANCEL"},
		{reactor.SignalTypeComplete, "COMPLETE"},
		{reactor.SignalTypeDefault, "DEFAULT"},
		{reactor.SignalTypeError, "ERROR"},
	} {
		assert.Equal(t, v.String, v.Signal.String(), "should be same string")
	}
	notExist := reactor.SignalType(0x7F)
	assert.Equal(t, "UNKNOWN", notExist.String())
}


<|start_filename|>hc/hc.go<|end_filename|>
package hc

import (
	"context"
	"io"
	"net/http"

	"github.com/jjeffcaii/reactor-go/mono"
)

// DefaultClient is the default Client.
var DefaultClient = NewClient(http.DefaultClient)

// Option is the type of http client options.
type Option func(*option)

// Mono is alias of mono.Mono.
type Mono = mono.Mono

// Client is a reactor style http client which returns a Mono.
type Client interface {
	// Do sends an HTTP request and returns a response Mono.
	Do(req *http.Request, opts ...Option) Mono
	// Get sends a GET request and returns a response Mono.
	Get(url string, opts ...Option) Mono
	// Delete sends a DELETE request and returns a response Mono.
	Delete(url string, opts ...Option) Mono
	// Post sends a POST request and returns a response Mono.
	Post(url string, contentType string, body io.Reader, opts ...Option) Mono
	// Put sends a PUT request and returns a response Mono.
	Put(url string, contentType string, body io.Reader, opts ...Option) Mono
	// Patch sends a PATCH request and returns a response Mono.
	Patch(url string, contentType string, body io.Reader, opts ...Option) Mono
}

// Do sends an HTTP request and returns a response Mono.
func Do(req *http.Request, opts ...Option) Mono {
	return DefaultClient.Do(req, opts...)
}

// Get sends a GET request and returns a response Mono.
func Get(url string, opts ...Option) Mono {
	return DefaultClient.Get(url, opts...)
}

// Delete sends a DELETE request and returns a response Mono.
func Delete(url string, opts ...Option) Mono {
	return DefaultClient.Delete(url, opts...)
}

// Post sends a POST request and returns a response Mono.
func Post(url string, contentType string, body io.Reader, opts ...Option) Mono {
	return DefaultClient.Post(url, contentType, body, opts...)
}

// Put sends a PUT request and returns a response Mono.
func Put(url string, contentType string, body io.Reader, opts ...Option) Mono {
	return DefaultClient.Put(url, contentType, body, opts...)
}

// Patch sends a PATCH request and returns a response Mono.
func Patch(url string, contentType string, body io.Reader, opts ...Option) Mono {
	return DefaultClient.Patch(url, contentType, body, opts...)
}

// NewClient creates a new Client.
func NewClient(cli *http.Client, defaultOptions ...Option) Client {
	return &client{
		c:    cli,
		opts: defaultOptions,
	}
}

var _ Client = (*client)(nil)

type client struct {
	c    *http.Client
	opts []Option
}

func (h *client) Do(req *http.Request, opts ...Option) Mono {
	gen := func(ctx context.Context) (*http.Request, error) {
		return req.WithContext(ctx), nil
	}
	return h.execute(gen, opts...)
}

func (h *client) Get(url string, opts ...Option) Mono {
	return h.customExecute(http.MethodGet, url, "", nil, opts...)
}

func (h *client) Delete(url string, opts ...Option) Mono {
	return h.customExecute(http.MethodDelete, url, "", nil, opts...)
}

func (h *client) Post(url string, contentType string, body io.Reader, opts ...Option) Mono {
	return h.customExecute(http.MethodPost, url, contentType, body, opts...)
}

func (h *client) Put(url string, contentType string, body io.Reader, opts ...Option) Mono {
	return h.customExecute(http.MethodPut, url, contentType, body, opts...)
}

func (h *client) Patch(url string, contentType string, body io.Reader, opts ...Option) Mono {
	return h.customExecute(http.MethodPatch, url, contentType, body, opts...)
}

func (h *client) customExecute(method string, url string, contentType string, body io.Reader, opts ...Option) Mono {
	return h.execute(func(ctx context.Context) (*http.Request, error) {
		req, err := http.NewRequestWithContext(ctx, method, url, body)
		if err != nil {
			return nil, err
		}
		if len(contentType) > 0 {
			req.Header.Set("Content-Type", contentType)
		}
		return req, nil
	}, opts...)
}

func (h *client) execute(gen func(ctx context.Context) (*http.Request, error), opts ...Option) Mono {
	return mono.Create(func(ctx context.Context, s mono.Sink) {
		var (
			opt *option
			req *http.Request
			err error
		)

		// init option
		if len(opts)+len(h.opts) > 0 {
			opt = new(option)
			for _, it := range h.opts {
				it(opt)
			}
			for _, it := range opts {
				it(opt)
			}
		}

		req, err = gen(ctx)
		if err != nil {
			s.Error(err)
			return
		}

		if opt != nil && len(opt.reqChain) > 0 {
			for _, it := range opt.reqChain {
				it(req)
			}
		}

		var res *http.Response
		res, err = h.c.Do(req)
		if err != nil {
			s.Error(err)
			return
		}

		if opt == nil || opt.bodyParser == nil {
			s.Success(res)
			return
		}

		var body interface{}
		body, err = opt.bodyParser(res)
		if err != nil {
			s.Error(err)
			return
		}

		s.Success(body)
	})
}


<|start_filename|>hc/hc_test.go<|end_filename|>
package hc_test

import (
	"context"
	"io"
	"net/http"
	"testing"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/hc"
	"github.com/jjeffcaii/reactor-go/mono"
	"github.com/jjeffcaii/reactor-go/scheduler"
	"github.com/jjeffcaii/reactor-go/tuple"
	"github.com/stretchr/testify/assert"
)

var httpBinUrl = "https://httpbin.org/anything"

func TestClient_Get(t *testing.T) {
	v, err := hc.Get("http://ipecho.net", hc.WithStringResponse()).Block(context.Background())
	assert.NoError(t, err, "request failed")
	assert.NotEmpty(t, v.(string), "should not be empty")

	cli := hc.NewClient(http.DefaultClient, hc.WithStringResponse())

	v, err = mono.
		Zip(
			cli.Get("http://ipecho.net/plain").SubscribeOn(scheduler.Parallel()),
			cli.Get("http://ipecho.net/json").SubscribeOn(scheduler.Parallel()),
			cli.Get("http://ipecho.net/xml").SubscribeOn(scheduler.Parallel()),
		).
		Block(context.Background())
	assert.NoError(t, err)

	tup := v.(tuple.Tuple)

	tup.ForEachWithIndex(func(v reactor.Any, e error, i int) bool {
		t.Logf("#%02d: value=%v, error=%v", i, v, e)
		return true
	})

	_, err = tup.Last()
	assert.Error(t, err)
}

type HttpBin struct {
	Method  string            `json:"method"`
	Url     string            `json:"url"`
	Headers map[string]string `json:"headers"`
}

func TestWithJSONBody(t *testing.T) {
	fakeHeaderKey := "X-Fake-Header"
	fakeHeaderValue := "fake value"
	v, err := hc.DefaultClient.
		Get(httpBinUrl, hc.WithRequestHijack(func(req *http.Request) {
			req.Header.Set(fakeHeaderKey, fakeHeaderValue)
		}), hc.WithJSONResponse(func() interface{} {
			return &HttpBin{}
		})).
		Block(context.Background())
	assert.NoError(t, err)
	assert.Equal(t, http.MethodGet, v.(*HttpBin).Method)
	assert.Equal(t, fakeHeaderValue, v.(*HttpBin).Headers[fakeHeaderKey])
}

func TestRequests(t *testing.T) {
	validate := func(v interface{}, err error) {
		assert.NoError(t, err)
		_, ok := v.(*http.Response)
		assert.True(t, ok)
	}

	for _, next := range []func(string, string, io.Reader, ...hc.Option) mono.Mono{
		hc.Post, hc.Put, hc.Patch,
	} {
		v, err := next(httpBinUrl, "application/json", nil).Block(context.Background())
		validate(v, err)
	}
	validate(hc.Get(httpBinUrl).Block(context.Background()))
	validate(hc.Delete(httpBinUrl).Block(context.Background()))
}

func TestClient_Do(t *testing.T) {
	req, _ := http.NewRequest(http.MethodGet, httpBinUrl, nil)
	v, err := hc.Do(req, hc.WithStringResponse()).Block(context.Background())
	assert.NoError(t, err, "should not return error")
	assert.NotEmpty(t, v, "should not be empty")
}

func TestDo_Failed(t *testing.T) {
	req, _ := http.NewRequest(http.MethodGet, "http://127.0.0.1/not-exists-path", nil)
	_, err := hc.Do(req).Block(context.Background())
	assert.Error(t, err)
}


<|start_filename|>flux/flux_interval_test.go<|end_filename|>
package flux_test

import (
	"context"
	"sync/atomic"
	"testing"
	"time"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/flux"
	"github.com/stretchr/testify/assert"
)

func TestInterval(t *testing.T) {
	done := make(chan struct{})
	var amount int32
	flux.Interval(100*time.Millisecond).
		DoFinally(func(s reactor.SignalType) {
			close(done)
		}).
		Subscribe(context.Background(),
			reactor.OnNext(func(v Any) error {
				atomic.AddInt32(&amount, 1)
				return nil
			}),
			reactor.OnSubscribe(func(ctx context.Context, su reactor.Subscription) {
				su.Request(3)
			}),
		)
	<-done
	assert.Equal(t, 3, int(amount), "bad amount")
}


<|start_filename|>internal/buffer/unbounded_test.go<|end_filename|>
package buffer_test

import (
	"strings"
	"sync/atomic"
	"testing"

	"github.com/jjeffcaii/reactor-go/internal/buffer"
	"github.com/stretchr/testify/assert"
)

func TestNewUnbounded(t *testing.T) {
	u := buffer.NewUnbounded()

	go func() {
		assert.True(t, u.Put("foo"))
		assert.True(t, u.Put("bar"))
		assert.True(t, u.Put("qux"))

		u.Dispose()
		assert.False(t, u.Put("must failed"))
	}()

	done := make(chan struct{})
	var read []string

	go func() {
		defer close(done)
		for next := range u.Get() {
			read = append(read, next.(string))
			u.Load()
		}
	}()

	<-done

	assert.Equal(t, "foo,bar,qux", strings.Join(read, ","), "result doesn't match")
}

func TestEmptyUnbounded(t *testing.T) {
	u := buffer.NewUnbounded()

	done := make(chan struct{})
	cnt := new(int32)

	go func() {
		defer close(done)
		for range u.Get() {
			atomic.AddInt32(cnt, 1)
			u.Load()
		}
	}()

	go func() {
		u.Dispose()
	}()

	<-done

	assert.Zero(t, atomic.LoadInt32(cnt))
}


<|start_filename|>flux/flux_test.go<|end_filename|>
package flux_test

import (
	"context"
	"errors"
	"fmt"
	"sync/atomic"
	"testing"
	"time"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/flux"
	"github.com/jjeffcaii/reactor-go/hooks"
	"github.com/jjeffcaii/reactor-go/scheduler"
	"github.com/stretchr/testify/assert"
)

type Any = reactor.Any

func Example() {
	gen := func(ctx context.Context, sink flux.Sink) {
		for i := 0; i < 10; i++ {
			v := i
			sink.Next(v)
		}
		sink.Complete()
	}
	done := make(chan struct{})

	var su reactor.Subscription
	flux.Create(gen).
		Filter(func(i Any) bool {
			return i.(int)%2 == 0
		}).
		Map(func(i interface{}) (Any, error) {
			return fmt.Sprintf("#HELLO_%04d", i.(int)), nil
		}).
		SubscribeOn(scheduler.Elastic()).
		Subscribe(context.Background(),
			reactor.OnSubscribe(func(ctx context.Context, s reactor.Subscription) {
				su = s
				s.Request(1)
			}),
			reactor.OnNext(func(v Any) error {
				fmt.Println("next:", v)
				su.Request(1)
				return nil
			}),
			reactor.OnComplete(func() {
				close(done)
			}),
		)
	<-done
}

var fakeErr = errors.New("fake error")
var testData = []Any{1, 2, 3, 4, 5}

func init() {
	flux.InitBuffSize(flux.BuffSizeXS)
}

func TestSuite(t *testing.T) {
	all := make(map[string]func() flux.Flux)
	var instJust = flux.Just(testData...)
	var instCreate = flux.Create(func(ctx context.Context, sink flux.Sink) {
		for _, it := range testData {
			sink.Next(it)
		}
		sink.Complete()
	})
	var instRange = flux.Range(1, 5)
	all["Just"] = func() flux.Flux {
		return instJust
	}
	all["Create"] = func() flux.Flux {
		return instCreate
	}
	all["Range"] = func() flux.Flux {
		return instRange
	}
	all["Unicast"] = func() flux.Flux {
		vv := flux.NewUnicastProcessor()
		go func() {
			for _, it := range testData {
				vv.Next(it)
			}
			vv.Complete()
		}()
		return vv
	}
	for k, gen := range all {
		t.Run(fmt.Sprintf("%s_Request", k), func(t *testing.T) {
			testRequest(gen(), t)
		})
		t.Run(fmt.Sprintf("%s_Peek", k), func(t *testing.T) {
			testPeek(gen(), t)
		})
		t.Run(fmt.Sprintf("%s_Discard", k), func(t *testing.T) {
			testDiscard(gen(), t)
		})
		t.Run(fmt.Sprintf("%s_FilterRequest", k), func(t *testing.T) {
			testFilterRequest(gen(), t)
		})
		t.Run(fmt.Sprintf("%s_BlockLast", k), func(t *testing.T) {
			testBlockLast(gen(), t)
		})
		t.Run(fmt.Sprintf("%s_BlockFirst", k), func(t *testing.T) {
			testBlockFirst(gen(), t)
		})
		t.Run(fmt.Sprintf("%s_DoSubscribeOn", k), func(t *testing.T) {
			testDoOnSubscribe(gen(), t)
		})
	}
}

func testDoOnSubscribe(f flux.Flux, t *testing.T) {
	var su reactor.Subscription
	var got int32
	onNext := func(v Any) error {
		atomic.AddInt32(&got, 1)
		su.Request(1)
		return nil
	}
	onSubscribe := func(ctx context.Context, s reactor.Subscription) {
		su = s
		su.Request(1)
	}
	_, err := f.DoOnNext(onNext).DoOnSubscribe(onSubscribe).BlockLast(context.Background())
	assert.NoError(t, err)
	assert.Len(t, testData, int(atomic.LoadInt32(&got)), "bad len")
}

func testBlockLast(f flux.Flux, t *testing.T) {
	last, err := f.BlockLast(context.Background())
	assert.NoError(t, err, "block last failed")
	assert.Equal(t, testData[len(testData)-1], last, "value doesn't match")
}

func testFilterRequest(f flux.Flux, t *testing.T) {
	var s reactor.Subscription
	var totals, discards, nexts, requests, filter int
	done := make(chan struct{})
	f.
		DoFinally(func(s reactor.SignalType) {
			assert.Equal(t, reactor.SignalTypeComplete, s, "bad signal")
			close(done)
		}).
		Filter(func(i interface{}) (ok bool) {
			totals++
			ok = i.(int)&1 == 0
			if ok {
				filter++
			}
			return
		}).
		DoOnDiscard(func(v interface{}) {
			discards++
		}).
		DoOnNext(func(v interface{}) error {
			nexts++
			s.Request(1)
			return nil
		}).
		DoOnRequest(func(n int) {
			requests++
		}).
		Subscribe(context.Background(), reactor.OnSubscribe(func(ctx context.Context, su reactor.Subscription) {
			s = su
			s.Request(1)
		}))
	<-done
	assert.Equal(t, totals, discards+nexts, "bad discards+nexts")
	assert.Equal(t, filter, nexts, "bad nexts")
	assert.Equal(t, nexts+1, requests, "bad requests")
}

func testDiscard(f flux.Flux, t *testing.T) {
	var next, next2, discard []Any
	done := make(chan struct{})
	f.
		DoFinally(func(s reactor.SignalType) {
			close(done)
		}).
		DoOnNext(func(v Any) error {
			next = append(next, v)
			return nil
		}).
		Filter(func(i Any) bool {
			return i.(int) > 3
		}).
		DoOnNext(func(v Any) error {
			next2 = append(next2, v)
			return nil
		}).
		DoOnDiscard(func(i Any) {
			discard = append(discard, i)
		}).
		Subscribe(context.Background())
	<-done
	assert.Equal(t, testData, next, "bad next")
	assert.Equal(t, len(next), len(next2)+len(discard), "bad amount")
	assert.Equal(t, []Any{4, 5}, next2, "bad next2")
	assert.Equal(t, []Any{1, 2, 3}, discard, "bad discard")
}

func testBlockFirst(f flux.Flux, t *testing.T) {
	// TODO: race problem
	//var cancelled bool
	first, err := f.
		//DoOnCancel(func() {
		//	cancelled = true
		//}).
		BlockFirst(context.Background())
	assert.NoError(t, err, "block first failed")
	assert.Equal(t, testData[0], first, "bad first value")
	//assert.True(t, cancelled, "should call OnCancel")
}

func testPeek(f flux.Flux, t *testing.T) {
	var complete int
	var a, b []int
	var requests int
	var ss reactor.Subscription
	done := make(chan struct{})
	f.
		DoOnNext(func(v interface{}) error {
			a = append(a, v.(int))
			return nil
		}).
		DoOnRequest(func(n int) {
			requests++
		}).
		DoOnComplete(func() {
			complete++
		}).
		DoFinally(func(s reactor.SignalType) {
			close(done)
		}).
		Subscribe(context.Background(), reactor.OnSubscribe(func(ctx context.Context, su reactor.Subscription) {
			ss = su
			ss.Request(1)
		}), reactor.OnNext(func(v interface{}) error {
			b = append(b, v.(int))
			ss.Request(1)
			return nil
		}))
	<-done
	assert.Equal(t, b, a, "values doesn't match")
	assert.Equal(t, len(a)+1, requests, "bad requests")
	assert.Equal(t, 1, complete, "bad complete")
}

func testRequest(f flux.Flux, t *testing.T) {
	var nexts []Any
	var su reactor.Subscription
	done := make(chan struct{})
	f.
		DoFinally(func(s reactor.SignalType) {
			close(done)
		}).
		SubscribeOn(scheduler.Elastic()).
		Subscribe(context.Background(), reactor.OnSubscribe(func(ctx context.Context, s reactor.Subscription) {
			su = s
			su.Request(1)
		}), reactor.OnNext(func(v Any) error {
			nexts = append(nexts, v)
			su.Request(1)
			return nil
		}))
	<-done
	assert.Equal(t, testData, nexts, "bad results")
}

func TestEmpty(t *testing.T) {
	innerTestEmpty(t, flux.Just())
	innerTestEmpty(t, flux.Empty())
}

func innerTestEmpty(t *testing.T, empty flux.Flux) {
	completed := int32(0)
	empty.Subscribe(
		context.Background(),
		reactor.OnNext(func(v Any) error {
			assert.Fail(t, "should be unreachable")
			return nil
		}),
		reactor.OnComplete(func() {
			atomic.AddInt32(&completed, 1)
		}),
	)
	assert.Equal(t, int32(1), atomic.LoadInt32(&completed))
}

func TestCreateWithRequest(t *testing.T) {
	const totals = 20
	f := flux.Create(func(ctx context.Context, sink flux.Sink) {
		for i := 0; i < totals; i++ {
			sink.Next(i)
		}
		sink.Complete()
	})

	var processed int32

	su := make(chan reactor.Subscription, 1)

	sub := reactor.NewSubscriber(reactor.OnNext(func(v interface{}) error {
		fmt.Println("next:", v)
		processed++
		return nil
	}), reactor.OnSubscribe(func(ctx context.Context, s reactor.Subscription) {
		su <- s
		s.Request(1)
	}), reactor.OnComplete(func() {
		fmt.Println("complete")
	}))

	time.AfterFunc(100*time.Millisecond, func() {
		(<-su).Request(totals - 1)
	})

	f.SubscribeWith(context.Background(), sub)
	assert.Equal(t, totals, int(processed), "bad processed num")
}

func TestNextWithError(t *testing.T) {
	var err error
	flux.Just(1).
		Subscribe(context.Background(), reactor.OnNext(func(v Any) error {
			return fakeErr
		}), reactor.OnError(func(e error) {
			err = e
		}))
	assert.Equal(t, fakeErr, err)
}

func TestBlockToSlice(t *testing.T) {
	results := make([]int, 0)
	err := flux.Just(3, 2, 1).BlockToSlice(context.Background(), &results)
	assert.NoError(t, err, "should not return error")
	assert.Equal(t, []int{3, 2, 1}, results)

	err = flux.Just("bad element type").BlockToSlice(context.Background(), &results)
	assert.Error(t, err, "should return error")

	err = flux.Just(1).BlockToSlice(context.Background(), nil)
	assert.Error(t, err)
}

func TestSubscribeWithChan(t *testing.T) {
	valChan := make(chan int)
	errChan := make(chan error)
	flux.Just(testData...).
		DoFinally(func(s reactor.SignalType) {
			close(valChan)
			close(errChan)
		}).
		SubscribeOn(scheduler.Parallel()).
		SubscribeWithChan(context.Background(), valChan, errChan)

	var actualData []int
L:
	for {
		select {
		case value, ok := <-valChan:
			if !ok {
				break L
			}
			actualData = append(actualData, value)
		case _, ok := <-errChan:
			if !ok {
				break L
			}
			assert.Fail(t, "should never run here")
		}
	}
	assert.Len(t, actualData, len(testData))
	for i := 0; i < len(testData); i++ {
		assert.Equal(t, testData[i], actualData[i], "result doesn't match")
	}

}

func TestSubscribeWithChan_Broken(t *testing.T) {
	valChan := make(chan int, 1)
	errChan := make(chan error, 1)
	flux.Just("bad element type").
		SubscribeOn(scheduler.Parallel()).
		SubscribeWithChan(context.Background(), valChan, errChan)
	select {
	case <-valChan:
		assert.Fail(t, "should be unreachable")
	case err := <-errChan:
		assert.Error(t, err, "should return error")
	}
}

func TestError(t *testing.T) {
	mockErr := errors.New("this is a mock error")
	var sig reactor.SignalType
	var e1, e2 error
	var requested int
	flux.Error(mockErr).
		DoFinally(func(s reactor.SignalType) {
			sig = s
		}).
		DoOnRequest(func(n int) {
			requested = n
		}).
		DoOnError(func(e error) {
			e1 = e
		}).
		Subscribe(
			context.Background(),
			reactor.OnNext(func(v interface{}) error {
				assert.Fail(t, "should never run here")
				return nil
			}),
			reactor.OnError(func(e error) {
				e2 = e
			}),
		)
	assert.Equal(t, mockErr, e1, "bad doOnError")
	assert.Equal(t, mockErr, e2, "bad onError")
	assert.Equal(t, reactor.SignalTypeError, sig, "bad signal")
	assert.True(t, requested > 0, "no request")
}

func TestMapFromErrorFlux(t *testing.T) {
	_, err := flux.Error(fakeErr).
		Map(func(any Any) (Any, error) {
			assert.Fail(t, "should be unreachable")
			return 1234, nil
		}).
		BlockLast(context.Background())
	assert.Equal(t, fakeErr, err, "should return error")
}

func TestMapWithError(t *testing.T) {
	_, err := flux.Just(134).
		Map(func(_ Any) (Any, error) {
			return nil, fakeErr
		}).
		BlockLast(context.Background())
	assert.Equal(t, fakeErr, err, "should return error")
}

func TestDoOnNextWithError(t *testing.T) {
	var e1 error
	_, err := flux.Just(1).
		DoOnNext(func(v Any) error {
			return fakeErr
		}).
		DoOnError(func(e error) {
			e1 = e
		}).
		BlockLast(context.Background())
	assert.Equal(t, fakeErr, e1)
	assert.Equal(t, fakeErr, err)
}

func TestDelayElement(t *testing.T) {
	start := time.Now()
	_, err := flux.Just(1, 2, 3).DelayElement(100 * time.Millisecond).BlockLast(context.Background())
	assert.NoError(t, err)
	assert.Equal(t, 3, int(time.Since(start).Milliseconds()/100))

	start = time.Now()
	_, err = flux.Error(fakeErr).DelayElement(100 * time.Millisecond).BlockLast(context.Background())
	assert.Equal(t, fakeErr, err)
	assert.True(t, time.Since(start) < 100*time.Millisecond)
}

func TestBlockLast(t *testing.T) {
	var dropped []reactor.Any

	hooks.OnNextDrop(func(v reactor.Any) {
		dropped = append(dropped, v)
	})

	v, err := flux.Just(1, 2, 3, 4).BlockLast(context.Background())
	assert.NoError(t, err)
	assert.Equal(t, 4, v)
	assert.Len(t, dropped, 3)

	_, err = flux.
		Create(func(ctx context.Context, sink flux.Sink) {
			sink.Error(errors.New("fake error"))
		}).
		SubscribeOn(scheduler.Parallel()).
		BlockLast(context.Background())
	assert.Error(t, err)
}


<|start_filename|>flux/op_take_test.go<|end_filename|>
package flux_test

import (
	"context"
	"testing"
	"time"

	"github.com/jjeffcaii/reactor-go/flux"
	"github.com/stretchr/testify/assert"
)

func TestTake(t *testing.T) {
	var amount int
	var completed bool
	var cancelled bool
	_, err := flux.Interval(10 * time.Millisecond).
		Take(3).
		DoOnCancel(func() {
			cancelled = true
		}).
		DoOnComplete(func() {
			completed = true
		}).
		DoOnNext(func(v Any) error {
			amount++
			return nil
		}).
		BlockLast(context.Background())
	assert.NoError(t, err, "block last failed")
	assert.Equal(t, 3, amount, "bad amount ")
	assert.True(t, completed, "bad completed")
	assert.False(t, cancelled, "bad cancelled")
}

func TestTakeFromError(t *testing.T) {
	_, err := flux.Error(fakeErr).Take(3).BlockLast(context.Background())
	assert.Equal(t, fakeErr, err)
}


<|start_filename|>mono/switch_value_if_error_test.go<|end_filename|>
package mono_test

import (
	"context"
	"errors"
	"fmt"
	"testing"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/mono"
	"github.com/stretchr/testify/assert"
)

func TestSwitchValueIfError(t *testing.T) {
	v, err := mono.JustOneshot(15555).
		Map(func(any reactor.Any) (reactor.Any, error) {
			return any, nil
		}).
		Map(func(any reactor.Any) (reactor.Any, error) {
			return any, fmt.Errorf("trigger err")
		}).
		DoOnError(func(e error) {
		}).
		SwitchValueIfError(123).
		Map(func(any reactor.Any) (reactor.Any, error) {
			return any, nil
		}).
		Block(context.Background())
	assert.NoError(t, err, "err occurred")
	assert.Equal(t, 123, v.(int), "bad result")

	fakeErr := errors.New("fake error")
	v, err = mono.Error(fakeErr).SwitchValueIfError(1).Block(context.Background())
	assert.NoError(t, err)
	assert.Equal(t, 1, v)

	v, err = mono.Just(1).SwitchValueIfError(2).Block(context.Background())
	assert.NoError(t, err)
	assert.Equal(t, 1, v)
}


<|start_filename|>scheduler/immediate_test.go<|end_filename|>
package scheduler_test

import (
	"fmt"
	"testing"

	"github.com/jjeffcaii/reactor-go/scheduler"
	"github.com/stretchr/testify/assert"
)

func TestImmediate(t *testing.T) {
	fmt.Println(scheduler.Immediate().Name())
	defer scheduler.Immediate().Close()
	assert.NotPanics(t, func() {
		_ = scheduler.Immediate().Worker().Do(func() {
			// noop
		})
	})
}


<|start_filename|>mono/delay_element_test.go<|end_filename|>
package mono_test

import (
	"context"
	"testing"
	"time"

	"github.com/jjeffcaii/reactor-go/mono"
	"github.com/stretchr/testify/assert"
)

func TestDelayElement(t *testing.T) {
	delay := 100 * time.Millisecond
	start := time.Now()
	v, err := mono.Just(1).DelayElement(delay).Block(context.Background())
	assert.NoError(t, err, "should not return error")
	assert.Equal(t, 1, v, "value doesn't match")
	assert.True(t, int64(time.Since(start)) > int64(delay), "should greater than delay duration")

	start = time.Now()
	v, err = mono.Empty().DelayElement(delay).Block(context.Background())
	assert.NoError(t, err, "should not return error")
	assert.Nil(t, v, "should be nil if mono is empty")
	assert.True(t, int64(time.Since(start)) < int64(delay), "should less than delay duration")

	start = time.Now()
	_, err = mono.Error(fakeErr).DelayElement(delay).Block(context.Background())
	assert.Equal(t, fakeErr, err, "should be fake error")
	assert.True(t, int64(time.Since(start)) < int64(delay), "should less than delay duration")
}


<|start_filename|>flux/flux_bench_test.go<|end_filename|>
package flux_test

import (
	"context"
	"testing"

	"github.com/jjeffcaii/reactor-go/flux"
)

func BenchmarkBlockLast(b *testing.B) {
	f := flux.Just(1, 2, 3, 4)
	b.ResetTimer()
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			_, _ = f.BlockLast(context.Background())
		}
	})
}
