GH-187: Added test for reset throttling.

test/11system_event_test.go

@@ -3,8 +3,10 @@ package test
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
 	"testing"
 
+	"github.com/resgateio/resgate/server"
 	"github.com/resgateio/resgate/server/reserr"
 )
 
@@ -335,3 +337,49 @@ func TestSystemReset_MismatchingResourceTypeResponseOnCollection_LogsError(t *te
 		s.AssertErrorsLogged(t, 1)
 	})
 }
+
+func TestSystemReset_WithThrottle_ThrottlesRequests(t *testing.T) {
+	const subscriptionCount = 5
+	const resetThrottle = 3
+	runTest(t, func(s *Session) {
+		c := s.Connect()
+		// Get subscriptions
+		for i := 1; i <= subscriptionCount; i++ {
+			subscribeToCustomResource(t, s, c, fmt.Sprintf("test.model.%d", i), resource{
+				typ:  typeModel,
+				data: fmt.Sprintf(`{"id":%d}`, i),
+			})
+		}
+		// Send system reset
+		s.SystemEvent("reset", json.RawMessage(`{"resources":["test.>"]}`))
+		// Get throttled number of requests
+		mreqs := s.GetParallelRequests(t, resetThrottle)
+		requestCount := resetThrottle
+		// Assert no other requests are sent
+		for i := 1; i <= subscriptionCount; i++ {
+			c.AssertNoNATSRequest(t, fmt.Sprintf("test.model.%d", i))
+		}
+		// Respond to requests one by one
+		for len(mreqs) > 0 {
+			r := mreqs[0]
+			mreqs = mreqs[1:]
+			id := r.Subject[strings.LastIndexByte(r.Subject, '.')+1:]
+			r.RespondSuccess(json.RawMessage(`{"model":` + fmt.Sprintf(`{"id":%s}`, id) + `}`))
+			// If we still have remaining subscriptions not yet received
+			if requestCount < subscriptionCount {
+				// For each response, a new request should be sent.
+				req := s.GetRequest(t)
+				mreqs = append(mreqs, req)
+				requestCount++
+				// Assert no other requests are sent
+				for i := 1; i <= subscriptionCount; i++ {
+					c.AssertNoNATSRequest(t, fmt.Sprintf("test.model.%d", i))
+				}
+			}
+		}
+
+	}, func(c *server.Config) {
+		c.ResetThrottle = resetThrottle
+	})
+
+}