logger/enable topic

Author: jmaralo

backend/cmd/main.go

@@ -32,6 +32,7 @@ import (
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/broker"
 	connection_topic "github.com/HyperloopUPV-H8/h9-backend/pkg/broker/topics/connection"
 	data_topic "github.com/HyperloopUPV-H8/h9-backend/pkg/broker/topics/data"
+	logger_topic "github.com/HyperloopUPV-H8/h9-backend/pkg/broker/topics/logger"
 	order_topic "github.com/HyperloopUPV-H8/h9-backend/pkg/broker/topics/order"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/transport"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/transport/network/sniffer"
@@ -155,10 +156,12 @@ func main() {
 	defer dataTopic.Stop()
 	connectionTopic := connection_topic.NewUpdateTopic()
 	orderTopic := order_topic.NewSendTopic()
+	loggerTopic := logger_topic.NewEnableTopic()
 
 	broker.AddTopic(data_topic.UpdateName, dataTopic)
 	broker.AddTopic(connection_topic.UpdateName, connectionTopic)
 	broker.AddTopic(order_topic.SendName, orderTopic)
+	broker.AddTopic(logger_topic.EnableName, loggerTopic)
 
 	connections := make(chan *websocket.Client)
 	upgrader := websocket.NewUpgrader(connections)
@@ -280,6 +283,26 @@ func main() {
 				if err != nil {
 					fmt.Println("Error pushing record to logger: ", err)
 				}
+			case logger_topic.EnableName:
+				status, ok := push.(*logger_topic.Status)
+				if !ok {
+					trace.Error().Any("push", push).Msg("error casting push to enable")
+					fmt.Printf("Push Type: %v\n", push)
+					return
+				}
+
+				var err error
+				if status.Enable() {
+					err = loggerHandler.Start()
+				} else {
+					err = loggerHandler.Stop()
+				}
+
+				if err != nil {
+					status.Fulfill(!status.Enable())
+				} else {
+					status.Fulfill(status.Enable())
+				}
 			default:
 				fmt.Printf("unknow topic %s\n", push.Topic())
 			}

backend/pkg/broker/topics/logger/enable.go

@@ -1,15 +1,29 @@
 package data
 
 import (
+	"encoding/json"
+	"fmt"
+	"sync/atomic"
+
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/websocket"
 )
 
 const EnableName abstraction.BrokerTopic = "logger/enable"
+const ResponseName abstraction.BrokerTopic = "logger/response"
 
 type Enable struct {
-	pool *websocket.Pool
-	api  abstraction.BrokerAPI
+	isRunning *atomic.Bool
+	pool      *websocket.Pool
+	api       abstraction.BrokerAPI
+}
+
+func NewEnableTopic() *Enable {
+	enable := &Enable{
+		isRunning: &atomic.Bool{},
+	}
+	enable.isRunning.Store(false)
+	return enable
 }
 
 func (enable *Enable) Topic() abstraction.BrokerTopic {
@@ -24,10 +38,74 @@ func (enable *Enable) Pull(request abstraction.BrokerRequest) (abstraction.Broke
 	return nil, nil
 }
 
+func (enable *Enable) ClientMessage(id websocket.ClientId, message *websocket.Message) {
+	switch message.Topic {
+	case EnableName:
+		err := enable.handleToggle(id, message)
+		if err != nil {
+			fmt.Printf("error handling logger: %v\n", err)
+		}
+	}
+}
+
+func (enable *Enable) handleToggle(id websocket.ClientId, message *websocket.Message) error {
+	var request bool
+	err := json.Unmarshal(message.Payload, &request)
+	if err != nil {
+		return err
+	}
+
+	status := newStatus(request)
+	go enable.api.UserPush(status)
+
+	go func() {
+		enable.isRunning.Store(<-status.response)
+		enable.broadcastState()
+	}()
+	return nil
+}
+
+func (enable *Enable) broadcastState() error {
+	payload, err := json.Marshal(enable.isRunning.Load())
+	if err != nil {
+		return err
+	}
+
+	enable.pool.Broadcast(websocket.Message{
+		Topic:   ResponseName,
+		Payload: payload,
+	})
+	return nil
+}
+
 func (enable *Enable) SetPool(pool *websocket.Pool) {
 	enable.pool = pool
 }
 
 func (enable *Enable) SetAPI(api abstraction.BrokerAPI) {
 	enable.api = api
 }
+
+type Status struct {
+	request  bool
+	response chan bool
+}
+
+func newStatus(request bool) *Status {
+	return &Status{
+		request:  request,
+		response: make(chan bool),
+	}
+}
+
+func (status *Status) Topic() abstraction.BrokerTopic {
+	return EnableName
+}
+
+func (status *Status) Fulfill(response bool) {
+	status.response <- response
+}
+
+func (status *Status) Enable() bool {
+	return status.request
+}

backend/pkg/logger/logger.go

@@ -111,7 +111,7 @@ func (logger *Logger) Stop() error {
 	defer logger.subloggersLock.Unlock()
 
 	if !logger.running.CompareAndSwap(true, false) {
-		fmt.Printf("Logger already stopped")
+		fmt.Println("Logger already stopped")
 		return nil
 	}