Merge remote-tracking branch 'origin/develop'

jmaralo · jmaralo · commit 45c081ff7b66 · 2024-01-01T22:14:07.000+01:00
diff --git a/backend/cmd/main.go b/backend/cmd/main.go
@@ -3,6 +3,7 @@ package main
 import (
 	"bufio"
 	"encoding/binary"
+	"errors"
 	"flag"
 	"fmt"
 	"log"
@@ -197,7 +198,7 @@ func main() {
 					Timestamp: packet.Timestamp,
 				})
 
-				if err != nil {
+				if err != nil && !errors.Is(err, logger.ErrLoggerNotRunning{}) {
 					fmt.Println("Error pushing record to data logger: ", err)
 				}
 
@@ -214,7 +215,7 @@ func main() {
 					Timestamp: packet.Timestamp,
 				})
 
-				if err != nil {
+				if err != nil && !errors.Is(err, logger.ErrLoggerNotRunning{}) {
 					fmt.Println("Error pushing record to info logger: ", err)
 				}
 
@@ -236,7 +237,7 @@ func main() {
 					Timestamp: packet.Timestamp,
 				})
 
-				if err != nil {
+				if err != nil && !errors.Is(err, logger.ErrLoggerNotRunning{}) {
 					fmt.Println("Error pushing record to info logger: ", err)
 				}
 
@@ -253,7 +254,7 @@ func main() {
 					Timestamp: packet.Timestamp,
 				})
 
-				if err != nil {
+				if err != nil && !errors.Is(err, logger.ErrLoggerNotRunning{}) {
 					fmt.Println("Error pushing record to state logger: ", err)
 				}
 
@@ -299,7 +300,7 @@ func main() {
 					Timestamp: packet.Timestamp(),
 				})
 
-				if err != nil {
+				if err != nil && !errors.Is(err, logger.ErrLoggerNotRunning{}) {
 					fmt.Println("Error pushing record to logger: ", err)
 				}
 			case logger_topic.EnableName:
diff --git a/backend/pkg/broker/topics/data/update.go b/backend/pkg/broker/topics/data/update.go
@@ -114,7 +114,7 @@ func (update *Update) send() error {
 	}
 
 	for _, id := range flaged {
-		update.pool.Disconnect(id, ws.CloseInternalServerErr, err.Error())
+		update.pool.Disconnect(id, ws.CloseInternalServerErr, "client disconnected")
 		delete(update.subscribers, id)
 		fmt.Printf("podData/update unsubscribed %s\n", uuid.UUID(id).String())
 	}
diff --git a/backend/pkg/broker/topics/message/update.go b/backend/pkg/broker/topics/message/update.go
@@ -66,7 +66,7 @@ func (update *Update) Push(p abstraction.BrokerPush) error {
 	}
 
 	for _, id := range flagged {
-		update.pool.Disconnect(id, ws.CloseUnsupportedData, "unsupported topic")
+		update.pool.Disconnect(id, ws.CloseUnsupportedData, "client disconnected")
 		delete(update.subscribers, id)
 		fmt.Printf("message/update unsubscribed %s\n", uuid.UUID(id).String())
 	}
diff --git a/backend/pkg/logger/data/logger.go b/backend/pkg/logger/data/logger.go
@@ -109,7 +109,7 @@ func (sublogger *Logger) PushRecord(record abstraction.LoggerRecord) error {
 		file, ok := sublogger.valueFileSlice[valueName]
 		if !ok {
 			filename := path.Join(
-				"loggerHandler/data",
+				"logger/data",
 				fmt.Sprintf("data_%s", loggerHandler.Timestamp.Format(time.RFC3339)),
 				fmt.Sprintf("%s.csv", valueName),
 			)
@@ -137,10 +137,9 @@ func (sublogger *Logger) PushRecord(record abstraction.LoggerRecord) error {
 
 		err := writer.Write([]string{
 			timestamp.Format(time.RFC3339),
-			val,
 			dataRecord.From,
 			dataRecord.To,
-			dataRecord.Timestamp.Format(time.RFC3339),
+			val,
 		})
 		if err != nil {
 			writerErr = loggerHandler.ErrWritingFile{
diff --git a/backend/pkg/logger/errors.go b/backend/pkg/logger/errors.go
@@ -40,6 +40,12 @@ func (err ErrLoggerNotRunning) Error() string {
 	return fmt.Sprintf("Error %s logger not running at %s", err.Name, err.Timestamp.Format(time.RFC3339))
 }
 
+func (err ErrLoggerNotRunning) Is(other error) bool {
+	_, ok := other.(ErrLoggerNotRunning)
+	_, okPtr := other.(*ErrLoggerNotRunning)
+	return ok || okPtr
+}
+
 type ErrWrongRecordType struct {
 	Name      abstraction.LoggerName
 	Timestamp time.Time
diff --git a/backend/pkg/logger/logger.go b/backend/pkg/logger/logger.go
@@ -38,19 +38,19 @@ func (logger Logger) UpdateMessage(_ wsModels.Client, message wsModels.Message)
 	err := json.Unmarshal(message.Payload, &enable)
 	if err != nil {
 		if err != nil {
-			fmt.Printf(ErrStoppingLogger{
+			fmt.Println(ErrStoppingLogger{
 				Name:      Name,
 				Timestamp: time.Now(),
 			}.Error())
 		}
-		fmt.Printf("Error unmarshalling enable message")
+		fmt.Println("Error unmarshalling enable message")
 		return
 	}
 
 	if enable {
 		err := logger.Start()
 		if err != nil {
-			fmt.Printf(ErrStartingLogger{
+			fmt.Println(ErrStartingLogger{
 				Name:      Name,
 				Timestamp: time.Now(),
 			}.Error())
@@ -59,7 +59,7 @@ func (logger Logger) UpdateMessage(_ wsModels.Client, message wsModels.Message)
 	} else {
 		err := logger.Stop()
 		if err != nil {
-			fmt.Printf(ErrStoppingLogger{
+			fmt.Println(ErrStoppingLogger{
 				Name:      Name,
 				Timestamp: time.Now(),
 			}.Error())
@@ -108,7 +108,7 @@ func (logger *Logger) Start() error {
 	for _, key := range logger.subloggers {
 		err := key.Start()
 		if err != nil {
-			fmt.Printf(ErrStartingLogger{
+			fmt.Println(ErrStartingLogger{
 				Name:      Name,
 				Timestamp: time.Now(),
 			}.Error())
@@ -121,23 +121,12 @@ func (logger *Logger) Start() error {
 
 // PushRecord works as a proxy for the PushRecord method of the subloggers
 func (logger *Logger) PushRecord(record abstraction.LoggerRecord) error {
-	objectiveLogger := record.Name()
-
-	for name, logger := range logger.subloggers {
-		if name == objectiveLogger {
-			err := logger.PushRecord(record)
-			if err != nil {
-				return ErrPushingRecord{
-					Name:      Name,
-					Timestamp: time.Now(),
-					Inner:     err,
-				}
-			}
-			return nil
-		}
+	sublogger, ok := logger.subloggers[record.Name()]
+	if !ok {
+		return ErrLoggerNotFound{record.Name()}
 	}
 
-	return ErrLoggerNotFound{objectiveLogger}
+	return sublogger.PushRecord(record)
 }
 
 // PullRecord works as a proxy for the PullRecord method of the subloggers
diff --git a/backend/pkg/logger/messages/logger.go b/backend/pkg/logger/messages/logger.go
@@ -122,9 +122,11 @@ func (sublogger *Logger) PushRecord(record abstraction.LoggerRecord) error {
 	defer writer.Flush()
 
 	err := writer.Write([]string{
+		infoRecord.Timestamp.Format(time.RFC3339),
+		infoRecord.From,
+		infoRecord.To,
 		timestamp,
 		msg,
-		infoRecord.From, infoRecord.To, infoRecord.Timestamp.Format(time.RFC3339),
 	})
 	if err != nil {
 		writerErr = err
diff --git a/backend/pkg/logger/order/logger.go b/backend/pkg/logger/order/logger.go
@@ -51,6 +51,30 @@ func (sublogger *Logger) Start() error {
 		return nil
 	}
 
+	filename := path.Join(
+		"logger/order",
+		fmt.Sprintf("order_%s", logger.Timestamp.Format(time.RFC3339)),
+		"order.csv",
+	)
+	err := os.MkdirAll(path.Dir(filename), os.ModePerm)
+	if err != nil {
+		return logger.ErrCreatingAllDir{
+			Name:      Name,
+			Timestamp: time.Now(),
+			Path:      filename,
+		}
+	}
+
+	file, err := os.Create(filename)
+	if err != nil {
+		return logger.ErrCreatingFile{
+			Name:      Name,
+			Timestamp: time.Now(),
+			Inner:     err,
+		}
+	}
+	sublogger.writer = file
+
 	fmt.Println("Logger started")
 	return nil
 }
@@ -76,43 +100,17 @@ func (sublogger *Logger) PushRecord(record abstraction.LoggerRecord) error {
 	sublogger.fileLock.Lock()
 	defer sublogger.fileLock.Unlock()
 
-	if sublogger.writer == nil {
-		filename := path.Join(
-			"logger/order",
-			fmt.Sprintf("order_%s", logger.Timestamp.Format(time.RFC3339)),
-			"order.csv",
-		)
-		err := os.MkdirAll(path.Dir(filename), os.ModePerm)
-		if err != nil {
-			return logger.ErrCreatingAllDir{
-				Name:      Name,
-				Timestamp: time.Now(),
-				Path:      filename,
-			}
-		}
-
-		file, err := os.Create(filename)
-		if err != nil {
-			return logger.ErrCreatingFile{
-				Name:      Name,
-				Timestamp: time.Now(),
-				Inner:     err,
-			}
-		}
-		sublogger.writer = file
-	}
-
 	csvWriter := csv.NewWriter(sublogger.writer)
 	defer csvWriter.Flush()
 
 	timestamp := orderRecord.Packet.Timestamp().Format(time.RFC3339)
 	val := fmt.Sprint(orderRecord.Packet.GetValues())
 	err := csvWriter.Write([]string{
 		timestamp,
-		fmt.Sprint(orderRecord.Packet.Id()),
-		val,
 		orderRecord.From,
 		orderRecord.To,
+		fmt.Sprint(orderRecord.Packet.Id()),
+		val,
 		orderRecord.Timestamp.Format(time.RFC3339),
 	})
 	if err != nil {

Original file line number	Diff line number	Diff line change
`@@ -114,7 +114,7 @@ func (update *Update) send() error {`
`114`	`114`	`}`
`115`	`115`
`116`	`116`	`for _, id := range flaged {`
`117`		`- update.pool.Disconnect(id, ws.CloseInternalServerErr, err.Error())`
	`117`	`+ update.pool.Disconnect(id, ws.CloseInternalServerErr, "client disconnected")`
`118`	`118`	`delete(update.subscribers, id)`
`119`	`119`	`fmt.Printf("podData/update unsubscribed %s\n", uuid.UUID(id).String())`
`120`	`120`	`}`
Original file line number	Diff line number	Diff line change
`@@ -66,7 +66,7 @@ func (update *Update) Push(p abstraction.BrokerPush) error {`
`66`	`66`	`}`
`67`	`67`
`68`	`68`	`for _, id := range flagged {`
`69`		`- update.pool.Disconnect(id, ws.CloseUnsupportedData, "unsupported topic")`
	`69`	`+ update.pool.Disconnect(id, ws.CloseUnsupportedData, "client disconnected")`
`70`	`70`	`delete(update.subscribers, id)`
`71`	`71`	`fmt.Printf("message/update unsubscribed %s\n", uuid.UUID(id).String())`
`72`	`72`	`}`