review comments

Signed-off-by: alecholmez <[email protected]>

move back recompute

Signed-off-by: alecholmez <[email protected]>

Author: alecholmez

pkg/server/sotw/v3/server.go

@@ -85,7 +85,7 @@ func (s *server) process(str stream.Stream, reqCh <-chan *discovery.DiscoveryReq
 	lastDiscoveryResponses := map[string]lastDiscoveryResponse{}
 
 	// a collection of stack allocated watches per request type
-	watches := newWatches()
+	watches := newWatches(reqCh)
 
 	defer func() {
 		watches.Cancel()

pkg/server/sotw/v3/watches.go

@@ -17,8 +17,7 @@ type watches struct {
 }
 
 // newWatches creates and initializes watches.
-func newWatches() watches {
-	// deltaMuxedResponses needs a buffer to release go-routines populating it
+func newWatches(req <-chan *discovery.DiscoveryRequest) watches {
 	return watches{
 		responders: make(map[string]*watch, int(types.UnknownType)),
 		cases:      make([]reflect.SelectCase, 2), // We use 2 for the default computation here: ctx.Done() + reqCh.Recv()
@@ -32,7 +31,7 @@ func (w *watches) Cancel() {
 	}
 }
 
-// recomputeWatches will analyze the currently typed list of the known watches and increase the known list of dynamic channels if needed
+// recomputeWatches rebuilds the known list of dynamic channels if needed
 func (w *watches) RecomputeWatches(ctx context.Context, reqCh <-chan *discovery.DiscoveryRequest) {
 	newCases := []reflect.SelectCase{
 		{
@@ -45,7 +44,7 @@ func (w *watches) RecomputeWatches(ctx context.Context, reqCh <-chan *discovery.
 		},
 	}
 
-	index := 2
+	index := len(newCases)
 	for _, watch := range w.responders {
 		newCases = append(newCases, watch.selectCase)
 		watch.index = index

pkg/server/v3/server_test.go

@@ -372,72 +372,92 @@ func TestFetch(t *testing.T) {
 
 	s := server.NewServer(context.Background(), config, cb)
 	out, err := s.FetchEndpoints(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out == nil || err != nil)
+	assert.NotNil(t, out)
+	assert.NoError(t, err)
 
 	out, err = s.FetchClusters(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out == nil || err != nil)
+	assert.NotNil(t, out)
+	assert.NoError(t, err)
 
 	out, err = s.FetchRoutes(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out == nil || err != nil)
+	assert.NotNil(t, out)
+	assert.NoError(t, err)
 
 	out, err = s.FetchListeners(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out == nil || err != nil)
+	assert.NotNil(t, out)
+	assert.NoError(t, err)
 
 	out, err = s.FetchSecrets(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out == nil || err != nil)
+	assert.NotNil(t, out)
+	assert.NoError(t, err)
 
 	out, err = s.FetchRuntime(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out == nil || err != nil)
+	assert.NotNil(t, out)
+	assert.NoError(t, err)
 
 	// try again and expect empty results
 	out, err = s.FetchEndpoints(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchClusters(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchRoutes(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchListeners(context.Background(), &discovery.DiscoveryRequest{Node: node})
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	// try empty requests: not valid in a real gRPC server
 	out, err = s.FetchEndpoints(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchClusters(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchRoutes(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchListeners(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchSecrets(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchRuntime(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	// send error from callback
 	callbackError = true
 	out, err = s.FetchEndpoints(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchClusters(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchRoutes(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	out, err = s.FetchListeners(context.Background(), nil)
-	assert.False(t, out != nil || err == nil)
+	assert.Nil(t, out)
+	assert.Error(t, err)
 
 	// verify fetch callbacks
-	assert.True(t, requestCount == 10)
-	assert.True(t, responseCount == 6)
+	assert.Equal(t, requestCount, 10)
+	assert.Equal(t, responseCount, 6)
 }
 
 func TestSendError(t *testing.T) {
@@ -591,7 +611,7 @@ func TestCancellations(t *testing.T) {
 	s := server.NewServer(context.Background(), config, server.CallbackFuncs{})
 	err := s.StreamAggregatedResources(resp)
 	assert.NoError(t, err)
-	assert.True(t, config.watches == 0)
+	assert.Equal(t, config.watches, 0)
 }
 
 func TestOpaqueRequestsChannelMuxing(t *testing.T) {
@@ -609,7 +629,7 @@ func TestOpaqueRequestsChannelMuxing(t *testing.T) {
 	s := server.NewServer(context.Background(), config, server.CallbackFuncs{})
 	err := s.StreamAggregatedResources(resp)
 	assert.NoError(t, err)
-	assert.True(t, config.watches == 0)
+	assert.Equal(t, config.watches, 0)
 }
 
 func TestCallbackError(t *testing.T) {