Function TestConcurrentEvalDrainReady

cel/program_async_test.go:707–781

(t *testing.T)

}

func TestConcurrentEvalDrainReady(t *testing.T) {
	cases := []struct {
		name       string
		expr       string
		debounce   time.Duration
		want       ref.Val
		wantPasses int
		minElapsed time.Duration
	}{
		{
			name:       "timer_reset",
			expr:       `delayed_rpc("a", 10) + delayed_rpc("b", 25) + delayed_rpc("c", 200)`,
			debounce:   250 * time.Millisecond,
			want:       types.String("abc"),
			wantPasses: 2,
			minElapsed: 200 * time.Millisecond,
		},
		{
			name:       "timer_reset",
			expr:       `delayed_rpc("a", 10) + delayed_rpc("b", 25) + delayed_rpc("c", 200)`,
			debounce:   60 * time.Millisecond,
			want:       types.String("abc"),
			wantPasses: 3,
			minElapsed: 200 * time.Millisecond,
		},
		{
			name:       "timer_already_fired",
			expr:       `delayed_rpc("a", 5) + delayed_rpc("b", 30) + delayed_rpc("c", 200)`,
			debounce:   1 * time.Millisecond,
			want:       types.String("abc"),
			wantPasses: 4,
			minElapsed: 200 * time.Millisecond,
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			var evalPasses atomic.Int32
			opts := []any{
				cel.Function("delayed_rpc",
					cel.Overload("delayed_rpc_string_int", []*cel.Type{cel.StringType, cel.IntType}, cel.StringType,
						cel.AsyncBinding(func(ctx context.Context, args ...ref.Val) ref.Val {
							msg := string(args[0].(types.String))
							delayMs := time.Duration(int64(args[1].(types.Int))) * time.Millisecond
							time.Sleep(delayMs)
							return types.String(msg)
						}),
					),
				),
				cel.ConcurrentDrainStrategy(async.DrainReady(tc.debounce)),
				trackEvalPasses(&evalPasses),
			}

			start := time.Now()
			prg := mustProgram(t, tc.expr, opts...)
			res := awaitEval(t, prg, context.Background(), cel.NoVars())
			elapsed := time.Since(start)