Function testConnectableComposed

factory_connectable_test.go:265–300

(t *testing.T, obs Observable)

}

func testConnectableComposed(t *testing.T, obs Observable) {
	obs = obs.Map(func(_ context.Context, i interface{}) (interface{}, error) {
		return i.(int) + 1, nil
	}, WithPublishStrategy())

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	eg, _ := errgroup.WithContext(ctx)

	expected := []interface{}{2, 3, 4}

	nbConsumers := 3
	wg := sync.WaitGroup{}
	wg.Add(nbConsumers)
	// Before Connect() is called we create multiple observers
	// We check all observers receive the same items
	for i := 0; i < nbConsumers; i++ {
		eg.Go(func() error {
			observer := obs.Observe(WithContext(ctx))
			wg.Done()

			got, err := collect(ctx, observer)
			if err != nil {
				return err
			}
			if !reflect.DeepEqual(got, expected) {
				return fmt.Errorf("expected: %v, got: %v", expected, got)
			}
			return nil
		})
	}

	wg.Wait()
	obs.Connect(ctx)
	assert.NoError(t, eg.Wait())
}

func testConnectableWithoutConnect(t *testing.T, obs Observable) {
	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)