Method next

observable_operator.go:162–171

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

}

func (op *averageIntOperator) next(ctx context.Context, item Item, dst chan<- Item, operatorOptions operatorOptions) {
	switch v := item.V.(type) {
	default:
		Error(IllegalInputError{error: fmt.Sprintf("expected type: int, got: %t", item)}).SendContext(ctx, dst)
		operatorOptions.stop()
	case int:
		op.sum += v
		op.count++
	}
}

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