Method GroupBy

observable_operator.go:1305–1351

GroupBy divides an Observable into a set of Observables that each emit a different group of items from the original Observable, organized by key.

(length int, distribution func(Item) int, opts ...Option)

// GroupBy divides an Observable into a set of Observables that each emit a different group of items from the original Observable, organized by key.
func (o *ObservableImpl) GroupBy(length int, distribution func(Item) int, opts ...Option) Observable {
	option := parseOptions(opts...)
	ctx := option.buildContext(o.parent)

	s := make([]Item, length)
	chs := make([]chan Item, length)
	for i := 0; i < length; i++ {
		ch := option.buildChannel()
		chs[i] = ch
		s[i] = Of(&ObservableImpl{
			iterable: newChannelIterable(ch),
		})
	}

	go func() {
		observe := o.Observe(opts...)
		defer func() {
			for i := 0; i < length; i++ {
				close(chs[i])
			}
		}()

		for {
			select {
			case <-ctx.Done():
				return
			case item, ok := <-observe:
				if !ok {
					return
				}
				idx := distribution(item)
				if idx >= length {
					err := Error(IndexOutOfBoundError{error: fmt.Sprintf("index %d, length %d", idx, length)})
					for i := 0; i < length; i++ {
						err.SendContext(ctx, chs[i])
					}
					return
				}
				item.SendContext(ctx, chs[idx])
			}
		}
	}()

	return &ObservableImpl{
		iterable: newSliceIterable(s, opts...),
	}
}

// GroupedObservable is the observable type emitted by the GroupByDynamic operator.
type GroupedObservable struct {