Method ConcurrentEval

cel/program.go:504–639

ConcurrentEval implements the Program interface.

(ctx context.Context, input any)

// ConcurrentEval implements the Program interface.
func (p *prog) ConcurrentEval(ctx context.Context, input any) <-chan EvalResult {
	resCh := make(chan EvalResult, 1)
	if ctx == nil {
		resCh <- EvalResult{Err: errors.New("context can not be nil")}
		close(resCh)
		return resCh
	}

	go func() {
		defer close(resCh)
		// Ensure concurrent eval handles panic / recovery properly
		defer func() {
			if r := recover(); r != nil {
				switch t := r.(type) {
				case interpreter.EvalCancelledError:
					resCh <- EvalResult{Err: t}
				default:
					resCh <- EvalResult{Err: fmt.Errorf("internal error: %v", r)}
				}
			}
		}()

		frame, err := p.newAsyncFrame(ctx, input)
		if err != nil {
			resCh <- EvalResult{Err: err}
			return
		}
		defer frame.Close()

		// Completions are signaled to this channel as async calls finish. The asyncCallState
		// fan-in also selects on ctx.Done(), so the sender will not leak if this loop returns early.
		completions := make(chan int64, p.resolveCompletionBufferSize())
		frame.SetCompletions(completions)

		for {
			var out ref.Val
			var det *EvalDetails

			if p.observable != nil {
				det = &EvalDetails{}
				out = p.observable.ObserveExec(frame, func(observed any) {
					switch o := observed.(type) {
					case interpreter.EvalState:
						det.state = o
					case *interpreter.CostTracker:
						det.costTracker = o
					}
				})
			} else {
				out = p.interpretable.Exec(frame)
			}

			// Communicate errors quickly.
			if types.IsError(out) {
				var err error = out.(*types.Err)
				if errors.Is(err, interpreter.InterruptError{}) {
					err = fmt.Errorf("%w: %w", err, context.Cause(ctx))
				}