perf: Queries issued by the /histories endpoint run concurrently

topfreegames · Oct 3, 2022 · 978952e · 978952e
1 parent 657c260
commit 978952e
Show file tree

Hide file tree

Showing 3 changed files with 36 additions and 19 deletions.
diff --git a/app/histories.go b/app/histories.go
@@ -2,6 +2,7 @@ package app
 
 import (
 	"net/http"
+	"sync"
 
 	"github.com/topfreegames/mqtt-history/mongoclient"
 
@@ -30,22 +31,35 @@ func HistoriesHandler(app *App) func(c echo.Context) error {
 			return c.String(echo.ErrUnauthorized.Code, echo.ErrUnauthorized.Message)
 		}
 
-		// retrieve messages
 		messages := make([]*models.Message, 0)
 		if app.Defaults.MongoEnabled {
 			collection := app.Defaults.MongoMessagesCollection
-
+			var wg sync.WaitGroup
+			var mu sync.Mutex
+			// guarantees ordering in responses payload
+			topicsMessagesMap := make(map[string][]*models.Message, len(authorizedTopics))
+			for _, topic := range authorizedTopics {
+				wg.Add(1)
+				go func(topicsMessagesMap map[string][]*models.Message, topic string) {
+					topicMessages := mongoclient.GetMessages(
+						c,
+						mongoclient.QueryParameters{
+							Topic:      topic,
+							From:       from,
+							Limit:      limit,
+							Collection: collection,
+						},
+					)
+					mu.Lock()
+					topicsMessagesMap[topic] = topicMessages
+					mu.Unlock()
+					wg.Done()
+				}(topicsMessagesMap, topic)
+			}
+			wg.Wait()
+			// guarantees ordering in responses payload
 			for _, topic := range authorizedTopics {
-				topicMessages := mongoclient.GetMessages(
-					c,
-					mongoclient.QueryParameters{
-						Topic:      topic,
-						From:       from,
-						Limit:      limit,
-						Collection: collection,
-					},
-				)
-				messages = append(messages, topicMessages...)
+				messages = append(messages, topicsMessagesMap[topic]...)
 			}
 			return c.JSON(http.StatusOK, messages)
 		}

diff --git a/mongoclient/get_messages.go b/mongoclient/get_messages.go
@@ -18,6 +18,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 
 	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/ext"
 	"github.com/opentracing/opentracing-go/log"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
@@ -190,8 +191,10 @@ func getMessagesPlayerSupportFromCollection(
 		ctx,
 		"get_messages_player_support_from_collection",
 		opentracing.Tags{
-			"db.statement": statement,
-			"db.type":      "mongo",
+			string(ext.DBStatement): statement,
+			string(ext.DBType):      "mongo",
+			string(ext.DBInstance):  database,
+			string(ext.DBUser):      user,
 		},
 	)
 	defer span.Finish()

diff --git a/mongoclient/get_messages_v2.go b/mongoclient/get_messages_v2.go
@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/ext"
 	"github.com/opentracing/opentracing-go/log"
 	"github.com/topfreegames/mqtt-history/logger"
 	"github.com/topfreegames/mqtt-history/models"
@@ -82,11 +83,10 @@ func getMessagesFromCollection(
 		ctx,
 		"get_messages_from_collection",
 		opentracing.Tags{
-			"span.kind":    "client",
-			"db.statement": statement,
-			"db.type":      "mongo",
-			"db.instance":  database,
-			"db.user":      user,
+			string(ext.DBStatement): statement,
+			string(ext.DBType):      "mongo",
+			string(ext.DBInstance):  database,
+			string(ext.DBUser):      user,
 		},
 	)
 	defer span.Finish()