Merge pull request #83 from HyperloopUPV-H8/backend/client-broker

[backend] [PR 2] Missing broker topics

Author: jmaralo

backend/cmd/main.go

@@ -17,14 +17,10 @@ import (
 
 	blcuPackage "github.com/HyperloopUPV-H8/h9-backend/internal/blcu"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/common"
-	"github.com/HyperloopUPV-H8/h9-backend/internal/connection_transfer"
-	"github.com/HyperloopUPV-H8/h9-backend/internal/data_transfer"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/excel"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/excel/ade"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/excel/utils"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/info"
-	"github.com/HyperloopUPV-H8/h9-backend/internal/message_transfer"
-	"github.com/HyperloopUPV-H8/h9-backend/internal/order_transfer"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/pod_data"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/server"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/update_factory"
@@ -34,6 +30,9 @@ 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"
+	message_topic "github.com/HyperloopUPV-H8/h9-backend/pkg/broker/topics/message"
+	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"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/transport/network/tcp"
@@ -111,20 +110,11 @@ func main() {
 	}
 	config.Vehicle.Network.Interface = dev.Name
 
-	connectionTransfer := connection_transfer.New(config.Connections)
-
 	vehicleOrders, err := vehicle_models.NewVehicleOrders(podData.Boards, config.Excel.Parse.Global.BLCUAddressKey)
 	if err != nil {
 		trace.Fatal().Err(err).Msg("creating vehicleOrders")
 	}
 
-	// <--- data transfer --->
-	dataTransfer := data_transfer.New(config.DataTransfer)
-	go dataTransfer.Run()
-
-	// <--- message transfer --->
-	messageTransfer := message_transfer.New(config.Messages)
-
 	// <--- update factory --->
 	updateFactory := update_factory.NewFactory()
 
@@ -146,24 +136,30 @@ func main() {
 			idToBoard[packet.Id] = board.Name
 		}
 	}
-	orderTransfer, orderChannel := order_transfer.New(idToBoard)
 
 	// <--- blcu --->
 	var blcu blcuPackage.BLCU
 	blcuAddr, useBlcu := info.Addresses.Boards["BLCU"]
 
 	// <--- broker --->
 	broker := broker.New()
-	broker.SetAPI(&brokerAPI{
-		OnUserPush: func(push abstraction.BrokerPush) {},
-	})
 
 	dataTopic := data_topic.NewUpdateTopic(time.Second / 10)
 	defer dataTopic.Stop()
 	connectionTopic := connection_topic.NewUpdateTopic()
+	orderTopic := order_topic.NewSendTopic()
+	loggerTopic := logger_topic.NewEnableTopic()
+	boardIdToBoard := make(map[abstraction.BoardId]string)
+	for name, id := range info.BoardIds {
+		boardIdToBoard[abstraction.BoardId(id)] = name
+	}
+	messageTopic := message_topic.NewUpdateTopic(boardIdToBoard)
 
 	broker.AddTopic(data_topic.UpdateName, dataTopic)
 	broker.AddTopic(connection_topic.UpdateName, connectionTopic)
+	broker.AddTopic(order_topic.SendName, orderTopic)
+	broker.AddTopic(logger_topic.EnableName, loggerTopic)
+	broker.AddTopic(message_topic.UpdateName, messageTopic)
 
 	connections := make(chan *websocket.Client)
 	upgrader := websocket.NewUpgrader(connections)
@@ -206,7 +202,10 @@ func main() {
 				}
 
 			case *info_packet.Packet:
-				messageTransfer.SendMessage(p)
+				err := broker.Push(message_topic.Push(p))
+				if err != nil {
+					fmt.Println(err)
+				}
 
 				err = loggerHandler.PushRecord(&messages_logger.Record{
 					Packet:    p,
@@ -220,7 +219,10 @@ func main() {
 				}
 
 			case *protection.Packet:
-				messageTransfer.SendMessage(p)
+				err := broker.Push(message_topic.Push(p))
+				if err != nil {
+					fmt.Println(err)
+				}
 
 				newPacket := info_packet.NewPacket(p.Id())
 				newPacket.BoardId = p.BoardId
@@ -256,10 +258,9 @@ func main() {
 				}
 
 			case *order.Add:
-				orderTransfer.AddStateOrders(*p)
-
+				trace.Debug().Msg("adding order")
 			case *order.Remove:
-				orderTransfer.RemoveStateOrders(*p)
+				trace.Debug().Msg("removing order")
 			}
 		},
 
@@ -268,6 +269,65 @@ func main() {
 		},
 	})
 
+	// this is here because we need to use transport to send messages
+	broker.SetAPI(&brokerAPI{
+		OnUserPush: func(push abstraction.BrokerPush) {
+			switch push.Topic() {
+			case order_topic.SendName:
+				order, ok := push.(*order_topic.Order)
+				if !ok {
+					trace.Error().Any("push", push).Msg("error casting push to order")
+					return
+				}
+
+				packet, err := order.ToPacket()
+				if err != nil {
+					trace.Error().Any("order", order).Err(err).Msg("error converting order to packet")
+					return
+				}
+
+				err = transp.SendMessage(transport.NewPacketMessage(packet))
+				if err != nil {
+					trace.Error().Any("order", order).Err(err).Msg("error sending order")
+					return
+				}
+
+				err = loggerHandler.PushRecord(&order_logger.Record{
+					Packet:    packet,
+					From:      "backend",
+					To:        idToBoard[uint16(packet.Id())],
+					Timestamp: packet.Timestamp(),
+				})
+
+				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())
+			}
+		},
+	})
+
 	// Load and set packet decoder and encoder
 	decoder, encoder := getTransportDecEnc(info, podData)
 	transp.WithDecoder(decoder).WithEncoder(encoder)
@@ -311,27 +371,6 @@ func main() {
 	}
 	go transp.HandleSniffer(sniffer.New(source, &layers.LayerTypeEthernet))
 
-	// <--- order transfer --->
-	go func() {
-		for order := range orderChannel {
-			err := transp.SendMessage(transport.NewPacketMessage(&order))
-			if err != nil {
-				trace.Error().Any("order", order).Err(err).Msg("error sending order")
-			}
-
-			err = loggerHandler.PushRecord(&order_logger.Record{
-				Packet:    &order,
-				From:      "backend",
-				To:        idToBoard[uint16(order.Id())],
-				Timestamp: order.Timestamp(),
-			})
-
-			if err != nil {
-				fmt.Println("Error pushing record to logger: ", err)
-			}
-		}
-	}()
-
 	// <--- blcu --->
 	if useBlcu {
 		blcu = blcuPackage.NewBLCU(net.TCPAddr{
@@ -352,12 +391,6 @@ func main() {
 		websocketBroker.RegisterHandle(&blcu, config.BLCU.Topics.Upload, config.BLCU.Topics.Download)
 	}
 
-	websocketBroker.RegisterHandle(&connectionTransfer, config.Connections.UpdateTopic, "connection/update")
-	websocketBroker.RegisterHandle(&dataTransfer, "podData/update")
-	websocketBroker.RegisterHandle(loggerHandler, config.LoggerHandler.Topics.Enable)
-	websocketBroker.RegisterHandle(&messageTransfer, "message/update")
-	websocketBroker.RegisterHandle(&orderTransfer, config.Orders.SendTopic, "order/stateOrders")
-
 	uploadableBords := common.Filter(common.Keys(info.Addresses.Boards), func(item string) bool {
 		return item != config.Excel.Parse.Global.BLCUAddressKey
 	})
@@ -568,6 +601,8 @@ func getTransportDecEnc(info info.Info, podData pod_data.PodData) (*presentation
 	protectionDecoder := protection.NewDecoder()
 	protectionDecoder.SetSeverity(abstraction.PacketId(info.MessageIds.Warning), protection.SeverityWarning)
 	protectionDecoder.SetSeverity(abstraction.PacketId(info.MessageIds.Fault), protection.SeverityFault)
+	decoder.SetPacketDecoder(abstraction.PacketId(info.MessageIds.Warning), protectionDecoder)
+	decoder.SetPacketDecoder(abstraction.PacketId(info.MessageIds.Fault), protectionDecoder)
 
 	return decoder, encoder
 }

backend/internal/vehicle/models/order_data.go

@@ -121,6 +121,7 @@ func getField(m pod_data.Measurement) (any, error) {
 				Kind: BooleanKind,
 				Name: typedMeas.Name,
 			},
+			VarType: "bool",
 		}, nil
 	case pod_data.EnumMeasurement:
 		return EnumDescription{
@@ -130,6 +131,7 @@ func getField(m pod_data.Measurement) (any, error) {
 				Name: typedMeas.Name,
 			},
 			Options: typedMeas.Options,
+			VarType: "enum",
 		}, nil
 	default:
 		return struct{}{}, errors.New("unrecognized measurement type")
@@ -157,9 +159,11 @@ type NumericDescription struct {
 
 type BooleanDescription struct {
 	fieldDescription
+	VarType string `json:"type"`
 }
 
 type EnumDescription struct {
 	fieldDescription
 	Options []string `json:"options"`
+	VarType string   `json:"type"`
 }

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
+}