Function runFirstItem

observable.go:386–421

(ctx context.Context, f func(interface{}) int, notif chan Item, observe <-chan Item, next chan Item, operatorFactory func() operator, option Option, opts ...Option)

}

func runFirstItem(ctx context.Context, f func(interface{}) int, notif chan Item, observe <-chan Item, next chan Item, operatorFactory func() operator, option Option, opts ...Option) {
	go func() {
		op := operatorFactory()
		stopped := false
		operator := operatorOptions{
			stop: func() {
				if option.getErrorStrategy() == StopOnError {
					stopped = true
				}
			},
			resetIterable: func(newIterable Iterable) {
				observe = newIterable.Observe(opts...)
			},
		}

	loop:
		for !stopped {
			select {
			case <-ctx.Done():
				break loop
			case i, ok := <-observe:
				if !ok {
					break loop
				}
				if i.Error() {
					op.err(ctx, i, next, operator)
					i.SendContext(ctx, notif)
				} else {
					op.next(ctx, i, next, operator)
					Of(f(i.V)).SendContext(ctx, notif)
				}
			}
		}
		op.end(ctx, next)
	}()
}

func (o *ObservableImpl) serialize(parent context.Context, fromCh chan Item, identifier func(interface{}) int, opts ...Option) Observable {
	option := parseOptions(opts...)