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

[backend] [PR 1] Broker with data and connection topics

Author: jmaralo

backend/cmd/main.go

@@ -13,6 +13,7 @@ import (
 	"runtime"
 	"runtime/pprof"
 	"strings"
+	"time"
 
 	blcuPackage "github.com/HyperloopUPV-H8/h9-backend/internal/blcu"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/common"
@@ -30,6 +31,9 @@ import (
 	vehicle_models "github.com/HyperloopUPV-H8/h9-backend/internal/vehicle/models"
 	"github.com/HyperloopUPV-H8/h9-backend/internal/ws_handle"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
+	"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"
 	"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"
@@ -40,6 +44,7 @@ import (
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/transport/packet/protection"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/transport/packet/state"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/transport/presentation"
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/websocket"
 	"github.com/fatih/color"
 	"github.com/google/gopacket/layers"
 	"github.com/google/gopacket/pcap"
@@ -147,6 +152,24 @@ func main() {
 	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()
+
+	broker.AddTopic(data_topic.UpdateName, dataTopic)
+	broker.AddTopic(connection_topic.UpdateName, connectionTopic)
+
+	connections := make(chan *websocket.Client)
+	upgrader := websocket.NewUpgrader(connections)
+	pool := websocket.NewPool(connections)
+	broker.SetPool(pool)
+
 	// <--- transport --->
 	orders := make(map[abstraction.PacketId]struct{})
 	for _, board := range podData.Boards {
@@ -159,14 +182,17 @@ func main() {
 
 	transp := transport.NewTransport()
 
-	transp.SetAPI(&TransportAPI{
+	transp.SetAPI(&transportAPI{
 		OnNotification: func(notification abstraction.TransportNotification) {
 			packet := notification.(transport.PacketNotification)
 
 			switch p := packet.Packet.(type) {
 			case *data.Packet:
 				update := updateFactory.NewUpdate(p)
-				dataTransfer.Update(update)
+				err := broker.Push(data_topic.NewPush(&update))
+				if err != nil {
+					fmt.Println(err)
+				}
 
 				err = loggerHandler.PushRecord(&data_logger.Record{
 					Packet: p,
@@ -226,7 +252,7 @@ func main() {
 		},
 
 		OnConnectionUpdate: func(target abstraction.TransportTarget, isConnected bool) {
-			connectionTransfer.Update(string(target), isConnected)
+			connectionTopic.Push(connection_topic.NewConnection(string(target), isConnected))
 		},
 	})
 
@@ -327,7 +353,7 @@ func main() {
 		ProgramableBoards: uploadableBords,
 	}
 
-	serverHandler, err := server.New(&websocketBroker, endpointData, config.Server)
+	serverHandler, err := server.New(upgrader, endpointData, config.Server)
 	if err != nil {
 		trace.Fatal().Err(err).Msg("Error creating server")
 		panic(err)
@@ -542,16 +568,16 @@ func getOps(units utils.Units) data.ConversionDescriptor {
 	return output
 }
 
-type TransportAPI struct {
+type transportAPI struct {
 	OnNotification     func(abstraction.TransportNotification)
 	OnConnectionUpdate func(abstraction.TransportTarget, bool)
 }
 
-func (api *TransportAPI) Notification(notification abstraction.TransportNotification) {
+func (api *transportAPI) Notification(notification abstraction.TransportNotification) {
 	api.OnNotification(notification)
 }
 
-func (api *TransportAPI) ConnectionUpdate(target abstraction.TransportTarget, isConnected bool) {
+func (api *transportAPI) ConnectionUpdate(target abstraction.TransportTarget, isConnected bool) {
 	api.OnConnectionUpdate(target, isConnected)
 }
 
@@ -605,3 +631,11 @@ func getTCPFilter(addrs []net.IP, serverPort uint16, clientPort uint16) string {
 	filter := fmt.Sprintf("(%s) and (%s) and (%s) and (%s) and (%s) and (%s)", ports, notSynFinRst, notJustAck, nonZeroPayload, srcAddressesStr, dstAddressesStr)
 	return filter
 }
+
+type brokerAPI struct {
+	OnUserPush func(abstraction.BrokerPush)
+}
+
+func (api *brokerAPI) UserPush(push abstraction.BrokerPush) {
+	api.OnUserPush(push)
+}

backend/internal/server/handler.go

@@ -3,6 +3,8 @@ package server
 import (
 	"fmt"
 	"sync"
+
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/websocket"
 )
 
 type Handler struct {
@@ -11,15 +13,15 @@ type Handler struct {
 	servers  map[string]*WebServer
 }
 
-func New(connections ConnectionHandler, data EndpointData, config Config) (*Handler, error) {
+func New(upgrader *websocket.Upgrader, data EndpointData, config Config) (*Handler, error) {
 	handler := &Handler{
 		config:   config,
 		serverMx: &sync.Mutex{},
 		servers:  make(map[string]*WebServer, len(config)),
 	}
 
 	for name, serverConfig := range config {
-		server, err := NewWebServer(name, connections, data, serverConfig)
+		server, err := NewWebServer(name, upgrader, data, serverConfig)
 		if err != nil {
 			return nil, err
 		}

backend/internal/server/server.go

@@ -5,16 +5,16 @@ import (
 	"net/http"
 	"sync/atomic"
 
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/websocket"
 	"github.com/gorilla/mux"
-	"github.com/gorilla/websocket"
 )
 
 type WebServer struct {
-	name        string
-	router      *mux.Router
-	connHandler ConnectionHandler
-	connected   *atomic.Int32
-	config      ServerConfig
+	name      string
+	router    *mux.Router
+	upgrader  *websocket.Upgrader
+	connected *atomic.Int32
+	config    ServerConfig
 }
 
 func NoCacheMiddleware(next http.Handler) http.Handler {
@@ -25,13 +25,13 @@ func NoCacheMiddleware(next http.Handler) http.Handler {
 	})
 }
 
-func NewWebServer(name string, connectionHandle ConnectionHandler, staticData EndpointData, config ServerConfig) (*WebServer, error) {
+func NewWebServer(name string, upgrader *websocket.Upgrader, staticData EndpointData, config ServerConfig) (*WebServer, error) {
 	server := &WebServer{
-		name:        name,
-		router:      mux.NewRouter(),
-		connHandler: connectionHandle,
-		connected:   &atomic.Int32{},
-		config:      config,
+		name:      name,
+		router:    mux.NewRouter(),
+		upgrader:  upgrader,
+		connected: &atomic.Int32{},
+		config:    config,
 	}
 
 	headers := map[string]string{
@@ -53,10 +53,7 @@ func NewWebServer(name string, connectionHandle ConnectionHandler, staticData En
 		return nil, err
 	}
 
-	upgrader := &websocket.Upgrader{
-		CheckOrigin: func(r *http.Request) bool { return true },
-	}
-	server.serveWebsocket(config.Endpoints.Connections, upgrader, headers)
+	server.router.Handle(config.Endpoints.Connections, server.upgrader)
 	server.serveFiles(config.Endpoints.Files, config.StaticPath)
 
 	if err != nil {

backend/internal/server/websocket.go

@@ -0,0 +1,77 @@
+package broker
+
+import (
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/broker/topics"
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/websocket"
+)
+
+var _ abstraction.Broker = &Broker{}
+
+type Broker struct {
+	topics  map[abstraction.BrokerTopic]topics.Handler
+	clients *websocket.Pool
+	api     abstraction.BrokerAPI
+}
+
+func New() *Broker {
+	return &Broker{
+		topics: make(map[abstraction.BrokerTopic]topics.Handler),
+	}
+}
+
+func (broker *Broker) Push(push abstraction.BrokerPush) error {
+	topic, ok := broker.topics[push.Topic()]
+	if !ok {
+		return ErrTopicNotFound{
+			Topic: push.Topic(),
+		}
+	}
+
+	return topic.Push(push)
+}
+
+func (broker *Broker) Pull(request abstraction.BrokerRequest) (abstraction.BrokerResponse, error) {
+	topic, ok := broker.topics[request.Topic()]
+	if !ok {
+		return nil, ErrTopicNotFound{
+			Topic: request.Topic(),
+		}
+	}
+
+	return topic.Pull(request)
+}
+
+func (broker *Broker) SetAPI(api abstraction.BrokerAPI) {
+	broker.api = api
+	for _, topic := range broker.topics {
+		topic.SetAPI(broker)
+	}
+}
+
+func (broker *Broker) SetPool(pool *websocket.Pool) {
+	broker.clients = pool
+	pool.SetOnMessage(broker.onMessage)
+	for _, topic := range broker.topics {
+		topic.SetPool(pool)
+	}
+}
+
+func (broker *Broker) AddTopic(topic abstraction.BrokerTopic, handler topics.Handler) {
+	handler.SetAPI(broker)
+	handler.SetPool(broker.clients)
+	broker.topics[topic] = handler
+}
+
+func (broker *Broker) onMessage(id websocket.ClientId, message *websocket.Message) {
+	topic, ok := broker.topics[message.Topic]
+	if !ok {
+		return //TODO: handle error
+	}
+
+	topic.ClientMessage(id, message)
+}
+
+func (broker *Broker) UserPush(push abstraction.BrokerPush) {
+	broker.api.UserPush(push)
+}

backend/pkg/broker/broker.go

@@ -0,0 +1,13 @@
+package broker
+
+import (
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
+)
+
+type ErrTopicNotFound struct {
+	Topic abstraction.BrokerTopic
+}
+
+func (err ErrTopicNotFound) Error() string {
+	return "topic not found"
+}

backend/pkg/broker/errors.go

@@ -0,0 +1,33 @@
+package blcu
+
+import (
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/websocket"
+)
+
+const DownloadName abstraction.BrokerTopic = "blcu/download"
+
+type Download struct {
+	pool *websocket.Pool
+	api  abstraction.BrokerAPI
+}
+
+func (download *Download) Topic() abstraction.BrokerTopic {
+	return DownloadName
+}
+
+func (download *Download) Push(push abstraction.BrokerPush) error {
+	return nil
+}
+
+func (download *Download) Pull(request abstraction.BrokerRequest) (abstraction.BrokerResponse, error) {
+	return nil, nil
+}
+
+func (download *Download) SetPool(pool *websocket.Pool) {
+	download.pool = pool
+}
+
+func (download *Download) SetAPI(api abstraction.BrokerAPI) {
+	download.api = api
+}