refactor: Change mqtt Subject to Topic

Author: NeedleInAJayStack

ingester.go

@@ -12,7 +12,7 @@ type ingester struct {
 	valueEmitter valueEmitter
 
 	// Mutable
-	// Stores a list of subject names that have been subscribed to.
+	// Stores a list of topic names that have been subscribed to.
 	topics map[string]map[uuid.UUID]bool
 	mux    *sync.RWMutex
 }
@@ -43,9 +43,9 @@ func (i *ingester) refreshSubscriptions(recs []rec) {
 
 	toSubscribe := []topicAndId{}
 	for _, record := range recs {
-		topic, ok := record.Tags["mqttSubject"].(string)
+		topic, ok := record.Tags["mqttTopic"].(string)
 		if !ok {
-			log.Printf("Error asserting type for mqttSubject")
+			log.Printf("Error asserting type for mqttTopic")
 			continue
 		}
 

ingester_test.go

@@ -37,19 +37,19 @@ func (suite *IngesterTestSuite) TestIngester() {
 	rec1 := rec{
 		ID: uuid.New(),
 		Tags: map[string]interface{}{
-			"mqttSubject": "test",
+			"mqttTopic": "test",
 		},
 	}
 	rec2 := rec{
 		ID: uuid.New(),
 		Tags: map[string]interface{}{
-			"mqttSubject": "test",
+			"mqttTopic": "test",
 		},
 	}
 	rec3 := rec{
 		ID: uuid.New(),
 		Tags: map[string]interface{}{
-			"mqttSubject": "test2",
+			"mqttTopic": "test2",
 		},
 	}
 

main.go

@@ -109,9 +109,9 @@ func main() {
 		currentStore,
 		&valueEmitter,
 	)
-	recs, err := recStore.readRecs("mqttSubject")
+	recs, err := recStore.readRecs("mqttTopic")
 	if err != nil {
-		log.Fatalf("error getting mqttSubject points: %s", err)
+		log.Fatalf("error getting mqttTopic points: %s", err)
 	}
 	ingester.refreshSubscriptions(recs)