DiscoveryClient      = discovery.AggregatedDiscoveryService_StreamAggregatedResourcesClient
	DeltaDiscoveryClient = discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesClient
)

var connectionNumber = atomic.NewUint32(0)

// ResponseHandler handles a XDS response in the agent. These will not be forwarded to Envoy.
// Currently, all handlers function on a single resource per type, so the API only exposes one
// resource.

	"istio.io/istio/pkg/log"
	"istio.io/istio/pkg/slices"
	"istio.io/istio/pkg/test/util/assert"
)

func TestLazySerial(t *testing.T) {
	t.Run("retry", func(t *testing.T) {
		computations := atomic.NewInt32(0)
		l := NewWithRetry(func() (int32, error) {
			res := computations.Inc()
			if res > 2 {
				return res, nil
			}

			return res, fmt.Errorf("not yet")
		})
		res, err := l.Get()
		assert.Error(t, err)

func TestCRDWatcherRace(t *testing.T) {
	stop := test.NewStop(t)
	c := kube.NewFakeClient()
	ctl := c.CrdWatcher()
	vsCalls := atomic.NewInt32(0)

	// Race callback and CRD creation
	go func() {
		if ctl.KnownOrCallback(gvr.VirtualService, func(s <-chan struct{}) {
			assert.Equal(t, s, stop)
			// Happened async
			vsCalls.Inc()
		}) {

//
// Deprecated: use generic Set instead.
// new ways:
// s1 := Set[int32]{}
// s2 := New[int32]()
type Int32 map[int32]Empty

// NewInt32 creates a Int32 from a list of values.
func NewInt32(items ...int32) Int32 {
	return Int32(New[int32](items...))
}

// Int32KeySet creates a Int32 from a keys of a map[int32](? extends interface{}).

	"time"

	"go.uber.org/atomic"
	"k8s.io/apimachinery/pkg/types"

	"istio.io/istio/pkg/test/util/assert"
	"istio.io/istio/pkg/test/util/retry"
)

func TestQueue(t *testing.T) {
	handles := atomic.NewInt32(0)
	q := NewQueue("custom", WithReconciler(func(key types.NamespacedName) error {
		handles.Inc()
		return nil
	}))
	q.Add(types.NamespacedName{Name: "something"})
	stop := make(chan struct{})
	go q.Run(stop)

		client:         client,
		revision:       revision,
		remote:         remote,
		defaultWatcher: watcher,
		perRevision:    perRevision,
		ttl:            time.Second,
		cycle:          atomic.NewInt32(0),
		enabled:        true,
	}
	l.AddRunFunction(func(stop <-chan struct{}) {
		<-stop
	})
	for _, fn := range fns {
		l.AddRunFunction(fn)
	}
	stop := make(chan struct{})
	go l.Run(stop)

	Failures  *atomic.Int32

	mu sync.Mutex
}

func NewFakeAuthenticator(name string) *FakeAuthenticator {
	return &FakeAuthenticator{
		Name:      name,
		Successes: atomic.NewInt32(0),
		Failures:  atomic.NewInt32(0),
	}
}

func (f *FakeAuthenticator) Authenticate(authCtx AuthContext) (*Caller, error) {
	if authCtx.GrpcContext != nil {
		return f.authenticateGrpc(authCtx.GrpcContext)
	}

						}
					}
				}
			}
		})
	}
}

func TestWebhookPatchingQueue(t *testing.T) {
	success := atomic.NewInt32(0)
	retries := atomic.NewInt32(0)
	queue := newWebhookPatcherQueue(func(key types.NamespacedName) error {
		if key.Name == "conflict-for-ever" {
			retries.Inc()
			return errors.New("conflict error")
		}

	// is called to ensure our event is not suppressed.
	i *atomic.Int32
}

func NewRecomputeTrigger() *RecomputeTrigger {
	inner := NewStatic[int32](ptr.Of(int32(0)))
	return &RecomputeTrigger{inner: inner, i: atomic.NewInt32(0)}
}

// TriggerRecomputation tells all dependants to recompute
func (r *RecomputeTrigger) TriggerRecomputation() {
	v := r.i.Inc()
	r.inner.Set(ptr.Of(v))
}

					assert.NoError(t, listener.Listen())
				}
				if prev && !want {
					assert.NoError(t, listener.Close())
				}
				<-cont
				prev = want
			}
		}()

		eventNum := atomic.NewInt32(0)
		go tcpHealthChecker.PerformApplicationHealthCheck(func(event *ProbeEvent) {
			if int(eventNum.Load()) >= len(tcpHealthStatuses) {
				return
			}
			if event.Healthy != expectedTCPEvents[eventNum.Load()].Healthy {