Method next

observable_operator.go:758–770

(ctx context.Context, item Item, dst chan<- Item, operatorOptions operatorOptions)

}

func (op *distinctOperator) next(ctx context.Context, item Item, dst chan<- Item, operatorOptions operatorOptions) {
	key, err := op.apply(ctx, item.V)
	if err != nil {
		Error(err).SendContext(ctx, dst)
		operatorOptions.stop()
		return
	}
	_, ok := op.keyset[key]
	if !ok {
		item.SendContext(ctx, dst)
	}
	op.keyset[key] = nil
}

func (op *distinctOperator) err(ctx context.Context, item Item, dst chan<- Item, operatorOptions operatorOptions) {
	defaultErrorFuncOperator(ctx, item, dst, operatorOptions)