Function customObservableOperator

observable.go:104–121

(parent context.Context, f func(ctx context.Context, next chan Item, option Option, opts ...Option), opts ...Option)

}

func customObservableOperator(parent context.Context, f func(ctx context.Context, next chan Item, option Option, opts ...Option), opts ...Option) Observable {
	option := parseOptions(opts...)
	next := option.buildChannel()
	ctx := option.buildContext(parent)

	if option.isEagerObservation() {
		go f(ctx, next, option, opts...)
		return &ObservableImpl{iterable: newChannelIterable(next)}
	}

	return &ObservableImpl{
		iterable: newFactoryIterable(func(propagatedOptions ...Option) <-chan Item {
			mergedOptions := append(opts, propagatedOptions...)
			go f(ctx, next, option, mergedOptions...)
			return next
		}),
	}
}

type operator interface {
	next(ctx context.Context, item Item, dst chan<- Item, operatorOptions operatorOptions)