Method next

observable_operator.go:806–817

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

}

func (op *distinctUntilChangedOperator) 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
	}
	if op.current != key {
		item.SendContext(ctx, dst)
		op.current = key
	}
}

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