Track data received for each subscription

bbb-graphql-middleware/cmd/bbb-graphql-middleware/main.go

@@ -35,7 +35,7 @@ func main() {
 			hasuraConnections := common.GetActivitiesOverview()["__HasuraConnection"].Started
 			topMessages := make(map[string]common.ActivitiesOverviewObj)
 			for index, item := range common.GetActivitiesOverview() {
-				if strings.HasPrefix(index, "_") || item.Started > hasuraConnections*3 {
+				if strings.HasPrefix(index, "_") || item.Started > hasuraConnections*3 || item.DataReceived > hasuraConnections*5 {
 					topMessages[index] = item
 				}
 			}

bbb-graphql-middleware/internal/common/GlobalState.go

@@ -16,8 +16,9 @@ func GetUniqueID() string {
 }
 
 type ActivitiesOverviewObj struct {
-	Started   int64
-	Completed int64
+	Started      int64
+	Completed    int64
+	DataReceived int64
 }
 
 var activitiesOverview = make(map[string]ActivitiesOverviewObj)
@@ -29,8 +30,9 @@ func ActivitiesOverviewStarted(index string) {
 
 	if _, exists := activitiesOverview[index]; !exists {
 		activitiesOverview[index] = ActivitiesOverviewObj{
-			Started:   0,
-			Completed: 0,
+			Started:      0,
+			Completed:    0,
+			DataReceived: 0,
 		}
 	}
 
@@ -40,6 +42,18 @@ func ActivitiesOverviewStarted(index string) {
 	activitiesOverview[index] = updatedValues
 }
 
+func ActivitiesOverviewDataReceived(index string) {
+	activitiesOverviewMux.Lock()
+	defer activitiesOverviewMux.Unlock()
+
+	if updatedValues, exists := activitiesOverview[index]; exists {
+		updatedValues.DataReceived++
+
+		activitiesOverview[index] = updatedValues
+	}
+
+}
+
 func ActivitiesOverviewCompleted(index string) {
 	activitiesOverviewMux.Lock()
 	defer activitiesOverviewMux.Unlock()

bbb-graphql-middleware/internal/hascli/conn/reader/reader.go

@@ -65,6 +65,10 @@ func handleMessageReceivedFromHasura(hc *common.HasuraConnection, fromHasuraToBr
 				common.ActivitiesOverviewCompleted("_Sum-" + string(subscription.Type))
 			}
 
+			if messageType == "data" {
+				common.ActivitiesOverviewDataReceived(string(subscription.Type) + "-" + subscription.OperationName)
+			}
+
 			if messageType == "data" &&
 				subscription.Type == common.Subscription {
 				hasNoPreviousOccurrence := handleSubscriptionMessage(hc, messageMap, subscription, queryId)