Function TestLeak

observable_operator_random_test.go:19–122

10 ms TODO Keep enriching tests

(t *testing.T)

// TODO Keep enriching tests
func TestLeak(t *testing.T) {
	var (
		count  = 100
		fooErr = errors.New("")
	)

	observables := map[string]func(context.Context) Observable{
		"Amb": func(ctx context.Context) Observable {
			obs := FromChannel(make(chan Item), WithContext(ctx))
			return Amb([]Observable{obs}, WithContext(ctx))
		},
		"CombineLatest": func(ctx context.Context) Observable {
			return CombineLatest(func(i ...interface{}) interface{} {
				sum := 0
				for _, v := range i {
					if v == nil {
						continue
					}
					sum += v.(int)
				}
				return sum
			}, []Observable{
				Just(1, 2)(),
				Just(10, 11)(),
			})
		},
		"Concat": func(ctx context.Context) Observable {
			return Concat([]Observable{
				Just(1, 2, 3)(),
				Just(4, 5, 6)(),
			})
		},
		"FromChannel": func(ctx context.Context) Observable {
			return FromChannel(getChannel(ctx), WithContext(ctx))
		},
		"FromEventSource": func(ctx context.Context) Observable {
			return FromEventSource(getChannel(ctx), WithContext(ctx))
		},
	}

	actions := map[string]func(context.Context, Observable){
		"All": func(ctx context.Context, obs Observable) {
			obs.All(func(_ interface{}) bool {
				return true
			}, WithContext(ctx))
		},
		"Average": func(ctx context.Context, obs Observable) {
			obs.AverageInt(WithContext(ctx))
		},
		"BufferWithTime": func(ctx context.Context, obs Observable) {
			obs.BufferWithTime(WithDuration(time.Millisecond), WithContext(ctx))
		},
		"Connect": func(ctx context.Context, obs Observable) {
			obs.Connect(ctx)
		},
		"Contains": func(ctx context.Context, obs Observable) {
			obs.Contains(func(i interface{}) bool {
				return i == 2