Asynchronous function invocation support

cel/cel_test.go

@@ -15,11 +15,13 @@
 package cel
 
 import (
+	"context"
 	"fmt"
 	"io/ioutil"
 	"log"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/golang/protobuf/proto"
 	"github.com/google/cel-go/checker/decls"
@@ -32,6 +34,7 @@ import (
 	"github.com/google/cel-go/interpreter"
 	"github.com/google/cel-go/interpreter/functions"
 	"github.com/google/cel-go/parser"
+	"github.com/google/cel-go/test"
 
 	descpb "github.com/golang/protobuf/protoc-gen-go/descriptor"
 	proto2pb "github.com/google/cel-go/test/proto2pb"
@@ -876,3 +879,98 @@ func Test_CustomInterpreterDecorator(t *testing.T) {
 		t.Errorf("got %v as the last observed constant, wanted 1", lastConst)
 	}
 }
+
+func Test_AsyncExtension(t *testing.T) {
+	env, err := NewEnv(
+		Declarations(
+			decls.NewVar("x", decls.String),
+			decls.NewFunction("asyncEcho",
+				decls.NewOverload(
+					"async_echo_string",
+					[]*exprpb.Type{decls.String},
+					decls.String)),
+		),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	funcs := Functions(
+		&functions.Overload{
+			Operator: "asyncEcho",
+			Async:    test.FakeRPC(25 * time.Millisecond),
+		},
+		&functions.Overload{
+			Operator: "async_echo_string",
+			Async:    test.FakeRPC(25 * time.Millisecond),
+		},
+	)
+
+	tests := []struct {
+		expr      string
+		parseOnly bool
+		evalOpts  EvalOption
+		out       ref.Val
+	}{
+		{
+			expr: `asyncEcho(x)`,
+			out:  types.String("async echo success!"),
+		},
+		{
+			expr:      `asyncEcho(x)`,
+			parseOnly: true,
+			out:       types.String("async echo success!"),
+		},
+		{
+			expr:     `asyncEcho(x)`,
+			evalOpts: OptOptimize,
+			out:      types.String("async echo success!"),
+		},
+		{
+			expr:     `asyncEcho(x) == 'async echo success!'`,
+			evalOpts: OptOptimize | OptTrackState,
+			out:      types.True,
+		},
+		{
+			expr:     `asyncEcho(x) == 'async echo success!' || true`,
+			evalOpts: OptOptimize | OptTrackState,
+			out:      types.True,
+		},
+	}
+	for i, tst := range tests {
+		tc := tst
+		t.Run(fmt.Sprintf("%d", i), func(tt *testing.T) {
+			var ast *Ast
+			var iss *Issues
+			if tc.parseOnly {
+				ast, iss = env.Parse(tc.expr)
+			} else {
+				ast, iss = env.Compile(tc.expr)
+			}
+			if iss.Err() != nil {
+				tt.Fatal(iss.Err())
+			}
+			opts := []ProgramOption{funcs}
+			if tc.evalOpts != 0 {
+				opts = append(opts, EvalOptions(tc.evalOpts))
+			}
+			prg, err := env.AsyncProgram(ast, opts...)
+			if err != nil {
+				tt.Fatal(err)
+			}
+			ctx := context.TODO()
+			out, det, err := prg.AsyncEval(ctx, map[string]interface{}{
+				"x": "async echo",
+			})
+			if err != nil {
+				tt.Fatal(err)
+			}
+			if out.Equal(tc.out) != types.True {
+				tt.Errorf("got %v, wanted %v", out, tc.out)
+			}
+			if tc.evalOpts&OptTrackState == OptTrackState && det == nil {
+				tt.Error("details was nil, expected non-nil")
+			}
+		})
+	}
+
+}

cel/env.go

@@ -307,14 +307,13 @@ func (e *Env) ParseSource(src common.Source) (*Ast, *Issues) {
 
 // Program generates an evaluable instance of the Ast within the environment (Env).
 func (e *Env) Program(ast *Ast, opts ...ProgramOption) (Program, error) {
-	optSet := e.progOpts
-	if len(opts) != 0 {
-		mergedOpts := []ProgramOption{}
-		mergedOpts = append(mergedOpts, e.progOpts...)
-		mergedOpts = append(mergedOpts, opts...)
-		optSet = mergedOpts
-	}
-	return newProgram(e, ast, optSet)
+	return e.newProgram(ast, opts /* async= */, false)
+}
+
+// AsyncProgram generates an evaluable instance of the Ast with support for asynchronous extension
+// functions.
+func (e *Env) AsyncProgram(ast *Ast, opts ...ProgramOption) (AsyncProgram, error) {
+	return e.newProgram(ast, opts /* async= */, true)
 }
 
 // SetFeature sets the given feature flag, as enumerated in options.go.
@@ -427,8 +426,8 @@ func (i *Issues) Err() error {
 	if i == nil {
 		return nil
 	}
-	if len(i.errs.GetErrors()) > 0 {
-		return errors.New(i.errs.ToDisplayString())
+	if len(i.Errors()) > 0 {
+		return errors.New(i.String())
 	}
 	return nil
 }