Function observable

observable.go:130–201

(parent context.Context, iterable Iterable, operatorFactory func() operator, forceSeq, bypassGather bool, opts ...Option)

}

func observable(parent context.Context, iterable Iterable, operatorFactory func() operator, forceSeq, bypassGather bool, opts ...Option) Observable {
	option := parseOptions(opts...)
	parallel, _ := option.getPool()

	if option.isEagerObservation() {
		next := option.buildChannel()
		ctx := option.buildContext(parent)
		if forceSeq || !parallel {
			runSequential(ctx, next, iterable, operatorFactory, option, opts...)
		} else {
			runParallel(ctx, next, iterable.Observe(opts...), operatorFactory, bypassGather, option, opts...)
		}
		return &ObservableImpl{iterable: newChannelIterable(next)}
	}

	if forceSeq || !parallel {
		return &ObservableImpl{
			iterable: newFactoryIterable(func(propagatedOptions ...Option) <-chan Item {
				mergedOptions := append(opts, propagatedOptions...)
				option := parseOptions(mergedOptions...)

				next := option.buildChannel()
				ctx := option.buildContext(parent)
				runSequential(ctx, next, iterable, operatorFactory, option, mergedOptions...)
				return next
			}),
		}
	}

	if serialized, f := option.isSerialized(); serialized {
		firstItemIDCh := make(chan Item, 1)
		fromCh := make(chan Item, 1)
		obs := &ObservableImpl{
			iterable: newFactoryIterable(func(propagatedOptions ...Option) <-chan Item {
				mergedOptions := append(opts, propagatedOptions...)
				option := parseOptions(mergedOptions...)

				next := option.buildChannel()
				ctx := option.buildContext(parent)
				observe := iterable.Observe(opts...)
				go func() {
					select {
					case <-ctx.Done():
						return
					case firstItemID := <-firstItemIDCh:
						if firstItemID.Error() {
							firstItemID.SendContext(ctx, fromCh)
							return
						}
						Of(firstItemID.V.(int)).SendContext(ctx, fromCh)
						runParallel(ctx, next, observe, operatorFactory, bypassGather, option, mergedOptions...)
					}
				}()
				runFirstItem(ctx, f, firstItemIDCh, observe, next, operatorFactory, option, mergedOptions...)
				return next
			}),
		}
		return obs.serialize(parent, fromCh, f)