Merge pull request #127 from HyperloopUPV-H8/develop

Merge for Ethernet View v1.0.0

Author: jmaralo

.gitignore

@@ -6,3 +6,6 @@ secret.json
 
 # MacOS Files
 .DS_Store
+
+# JetBrains IDE
+.idea/

backend/cmd/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bufio"
+	"context"
 	"encoding/binary"
 	"flag"
 	"fmt"
@@ -57,6 +58,7 @@ import (
 var traceLevel = flag.String("trace", "info", "set the trace level (\"fatal\", \"error\", \"warn\", \"info\", \"debug\", \"trace\")")
 var traceFile = flag.String("log", "trace.json", "set the trace log file")
 var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")
+var enableSNTP = flag.Bool("sntp", false, "enables a simple SNTP server on port 123")
 
 func main() {
 	flag.Parse()
@@ -131,7 +133,7 @@ func main() {
 		state_logger.Name:      state_logger.NewLogger(),
 	}
 
-	loggerHandler := logger.NewLogger(subloggers)
+	loggerHandler := logger.NewLogger(subloggers, trace.Logger)
 
 	// <--- order transfer --->
 	idToBoard := make(map[uint16]string)
@@ -142,7 +144,7 @@ func main() {
 	}
 
 	// <--- broker --->
-	broker := broker.New()
+	broker := broker.New(trace.Logger)
 
 	dataTopic := data_topic.NewUpdateTopic(time.Second / 10)
 	defer dataTopic.Stop()
@@ -162,20 +164,20 @@ func main() {
 	broker.AddTopic(message_topic.UpdateName, messageTopic)
 
 	connections := make(chan *websocket.Client)
-	upgrader := websocket.NewUpgrader(connections)
-	pool := websocket.NewPool(connections)
+	upgrader := websocket.NewUpgrader(connections, trace.Logger)
+	pool := websocket.NewPool(connections, trace.Logger)
 	broker.SetPool(pool)
 
 	// <--- transport --->
-	transp := transport.NewTransport()
+	transp := transport.NewTransport(trace.Logger)
 
 	// <--- vehicle --->
 	ipToBoardId := make(map[string]abstraction.BoardId)
 	for name, ip := range info.Addresses.Boards {
 		ipToBoardId[ip.String()] = abstraction.BoardId(info.BoardIds[name])
 	}
 
-	vehicle := vehicle.New()
+	vehicle := vehicle.New(trace.Logger)
 	vehicle.SetBroker(broker)
 	vehicle.SetLogger(loggerHandler)
 	vehicle.SetUpdateFactory(updateFactory)
@@ -193,6 +195,7 @@ func main() {
 		for _, packet := range board.Packets {
 			transp.SetIdTarget(abstraction.PacketId(packet.Id), abstraction.TransportTarget(board.Name))
 		}
+		transp.SetTargetIp(info.Addresses.Boards[board.Name].String(), abstraction.TransportTarget(board.Name))
 	}
 
 	// Start handling TCP client connections
@@ -207,12 +210,17 @@ func main() {
 		if err != nil {
 			panic("Failed to resolve local backend TCP client address")
 		}
-		go transp.HandleClient(tcp.NewClient(backendTcpClientAddr), abstraction.TransportTarget(board.Name), "tcp", fmt.Sprintf("%s:%d", info.Addresses.Boards[board.Name], info.Ports.TcpServer))
+		go transp.HandleClient(tcp.NewClientConfig(backendTcpClientAddr), fmt.Sprintf("%s:%d", info.Addresses.Boards[board.Name], info.Ports.TcpServer))
 		i++
 	}
 
 	// Start handling TCP server connections
-	go transp.HandleServer(tcp.NewServer(serverTargets), "tcp", fmt.Sprintf("%s:%d", info.Addresses.Backend, info.Ports.TcpServer))
+	go transp.HandleServer(tcp.ServerConfig{
+		ListenConfig: net.ListenConfig{
+			KeepAlive: time.Second,
+		},
+		Context: context.TODO(),
+	}, fmt.Sprintf("%s:%d", info.Addresses.Backend, info.Ports.TcpServer))
 
 	// Start handling the sniffer
 	source, err := pcap.OpenLive(dev.Name, 1500, true, pcap.BlockForever)
@@ -227,7 +235,7 @@ func main() {
 	if err != nil {
 		panic("failed to compile bpf filter")
 	}
-	go transp.HandleSniffer(sniffer.New(source, &layers.LayerTypeEthernet))
+	go transp.HandleSniffer(sniffer.New(source, &layers.LayerTypeEthernet, trace.Logger))
 
 	// <--- http server --->
 	podDataHandle, err := h.HandleDataJSON("podData.json", pod_data.GetDataOnlyPodData(podData))
@@ -260,24 +268,26 @@ func main() {
 	}
 
 	// <--- SNTP --->
-	sntpAddr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", info.Addresses.Backend, info.Ports.SNTP))
-	if err != nil {
-		fmt.Fprintf(os.Stderr, "error resolving sntp address: %v\n", err)
-		os.Exit(1)
-	}
-	sntpServer, err := sntp.NewUnicast("udp", sntpAddr)
-	if err != nil {
-		fmt.Fprintf(os.Stderr, "error creating sntp server: %v\n", err)
-		os.Exit(1)
-	}
-
-	go func() {
-		err := sntpServer.ListenAndServe()
+	if *enableSNTP {
+		sntpAddr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", info.Addresses.Backend, info.Ports.SNTP))
 		if err != nil {
-			fmt.Fprintf(os.Stderr, "error listening sntp server: %v\n", err)
-			return
+			fmt.Fprintf(os.Stderr, "error resolving sntp address: %v\n", err)
+			os.Exit(1)
 		}
-	}()
+		sntpServer, err := sntp.NewUnicast("udp", sntpAddr)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "error creating sntp server: %v\n", err)
+			os.Exit(1)
+		}
+
+		go func() {
+			err := sntpServer.ListenAndServe()
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "error listening sntp server: %v\n", err)
+				return
+			}
+		}()
+	}
 
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt)
@@ -390,8 +400,8 @@ func getConfig(path string) Config {
 }
 
 func getTransportDecEnc(info info.Info, podData pod_data.PodData) (*presentation.Decoder, *presentation.Encoder) {
-	decoder := presentation.NewDecoder(binary.LittleEndian)
-	encoder := presentation.NewEncoder(binary.LittleEndian)
+	decoder := presentation.NewDecoder(binary.LittleEndian, trace.Logger)
+	encoder := presentation.NewEncoder(binary.LittleEndian, trace.Logger)
 
 	dataDecoder := data.NewDecoder(binary.LittleEndian)
 	dataEncoder := data.NewEncoder(binary.LittleEndian)

backend/cmd/trace.go

@@ -21,14 +21,6 @@ var traceLevelMap = map[string]zerolog.Level{
 
 func initTrace(traceLevel string, traceFile string) *os.File {
 	zerolog.CallerMarshalFunc = func(pc uintptr, file string, line int) string {
-		short := file
-		for i := len(file) - 1; i > 0; i-- {
-			if file[i] == '/' {
-				short = file[i+1:]
-				break
-			}
-		}
-		file = short
 		return file + ":" + strconv.Itoa(line)
 	}
 	zerolog.TimeFieldFormat = zerolog.TimeFormatUnixNano
@@ -45,8 +37,7 @@ func initTrace(traceLevel string, traceFile string) *os.File {
 
 	multi := zerolog.MultiLevelWriter(consoleWriter, file)
 
-	global_logger := zerolog.New(multi).With().Timestamp().Caller().Logger()
-	trace.Logger = global_logger
+	trace.Logger = zerolog.New(multi).With().Timestamp().Caller().Logger()
 
 	if level, ok := traceLevelMap[traceLevel]; ok {
 		zerolog.SetGlobalLevel(level)

backend/pkg/abstraction/broker.go

@@ -32,9 +32,11 @@ type Broker interface {
 	SetAPI(BrokerAPI)
 }
 
-// BrokerAPi is the API provided for the Broker by the Vehicle.
+// BrokerAPI is the API provided for the Broker by the Vehicle.
 // The Broker must use this to communicate with the rest of the code
 type BrokerAPI interface {
-	// UserPush notifies that the front-end has sent information without a previous request
-	UserPush(BrokerPush)
+	// UserPush notifies that the front-end has sent information without a previous request.
+	UserPush(BrokerPush) error
+	// UserPull notifies that the front-end has requested information.
+	UserPull(BrokerRequest) (BrokerResponse, error)
 }