Method ZipFromIterable

observable_operator.go:2960–3010

ZipFromIterable merges the emissions of an Iterable via a specified function and emit single items for each combination based on the results of this function.

(iterable Iterable, zipper Func2, opts ...Option)

// ZipFromIterable merges the emissions of an Iterable via a specified function
// and emit single items for each combination based on the results of this function.
func (o *ObservableImpl) ZipFromIterable(iterable Iterable, zipper Func2, opts ...Option) Observable {
	option := parseOptions(opts...)
	next := option.buildChannel()
	ctx := option.buildContext(o.parent)

	go func() {
		defer close(next)
		it1 := o.Observe(opts...)
		it2 := iterable.Observe(opts...)
	loop:
		for {
			select {
			case <-ctx.Done():
				break loop
			case i1, ok := <-it1:
				if !ok {
					break loop
				}
				if i1.Error() {
					i1.SendContext(ctx, next)
					return
				}
				for {
					select {
					case <-ctx.Done():
						break loop
					case i2, ok := <-it2:
						if !ok {
							break loop
						}
						if i2.Error() {
							i2.SendContext(ctx, next)
							return
						}
						v, err := zipper(ctx, i1.V, i2.V)
						if err != nil {
							Error(err).SendContext(ctx, next)
							return
						}
						Of(v).SendContext(ctx, next)
						continue loop
					}
				}
			}
		}
	}()

	return &ObservableImpl{
		iterable: newChannelIterable(next),
	}
}