Alignement

README.md

@@ -12,7 +12,41 @@ without having to be playing the game while doing it (fans are noisy).
 
 # Usage
 To start replaying:
-`BeamNGMockOg replay -a 127.0.0.1 -p 4443 -i sunburstManual.bin`
+`BeamNGMockOg replay [--loop] -a 127.0.0.1 -p 4443 -i sunburstManual.bin`
+- `loop` allows the replay functionality to keep replaying the same data file
 
 To start recording:
 `BeamNGMockOg record -a 127.0.0.1 -p 4443 -o sunburstDCT.bin`
+
+## Development
+Use `tcpdump` to listen to the socket and check if data is coming through:
+`tcpdump -i any udp port <port> -X`
+
+### Benchmark
+Run with:
+`go test ./mockserver -bench=BenchmarkReplayAsync -benchmem -benchtime=1x -memprofile=mem.pprof`
+
+Analyze the output with:
+`go tool pprof -sample_index=alloc_objects mem.pprof` -> `top`
+- `ignore=net` will ignore the net package for example
+- `focus=mockserver` will show only the results from this code we are testing
+or
+`go tool pprof -http=:8080 mem.pprof`
+
+#### Previous results:
+goos: linux
+goarch: amd64
+pkg: github.com/ESilva15/BeamNGMockOg/mockserver
+cpu: AMD Ryzen 5 5600G with Radeon Graphics
+BenchmarkReplayAsync-12    	      1	109433574741 ns/op	4533008 B/op	  66059 allocs/op
+PASS
+ok  	github.com/ESilva15/BeamNGMockOg/mockserver	109.438s
+
+#### Current results
+goos: linux
+goarch: amd64
+pkg: github.com/ESilva15/BeamNGMockOg/mockserver
+cpu: AMD Ryzen 5 5600G with Radeon Graphics
+BenchmarkReplayAsync-12    	      1	109433931841 ns/op	 356088 B/op	  13272 allocs/op
+PASS
+ok  	github.com/ESilva15/BeamNGMockOg/mockserver	109.440s

cmd/replay.go

@@ -15,8 +15,14 @@ func replayAction(cmd *cobra.Command, args []string) {
 	port, _ := cmd.Flags().GetInt("port")
 	loop, _ := cmd.Flags().GetBool("loop")
 
+	replayer, err := mockserver.NewReplayer(address, port, inputFile)
+	if err != nil {
+		fmt.Printf("Something went wrong setting up the player: %+v", err)
+		return
+	}
+
 	ctx := context.Background()
-	if err := mockserver.Replay(ctx, address, port, loop, inputFile); err != nil {
+	if err := replayer.Replay(ctx, loop); err != nil {
 		fmt.Printf("Something went wrong while playing the file: %v", err)
 	}
 }

mockserver/reader.go

@@ -1,25 +1,23 @@
 package mockserver
 
 import (
-	"bytes"
-	"encoding/binary"
-	"encoding/gob"
 	"io"
+	"unsafe"
 
 	sdk "github.com/ESilva15/gobngsdk"
 )
 
 type GobReader struct {
-	TotalRead int
+	TotalRead int64
 	File      io.ReadSeeker
-	Dec       *gob.Decoder
+	Buf       []byte
 }
 
 func NewGobReader(r io.ReadSeeker) *GobReader {
 	return &GobReader{
 		TotalRead: 0,
 		File:      r,
-		Dec:       gob.NewDecoder(r),
+		Buf:       make([]byte, unsafe.Sizeof(sdk.Outgauge{})),
 	}
 }
 
@@ -29,27 +27,19 @@ func (g *GobReader) Reset() error {
 		return err
 	}
 
-	g.Dec = gob.NewDecoder(g.File)
 	g.TotalRead = 0
 
 	return nil
 }
 
-func (g *GobReader) Next() ([]byte, error) {
-	var og sdk.Outgauge
-
-	err := g.Dec.Decode(&og)
+func (g *GobReader) Next(buffer []byte) error {
+	_, err := io.ReadFull(g.File, buffer)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	buf := new(bytes.Buffer)
-	err = binary.Write(buf, binary.LittleEndian, &og)
-	if err != nil {
-		return nil, err
-	}
+	pos, _ := g.File.Seek(0, io.SeekCurrent)
+	g.TotalRead = pos
 
-	g.TotalRead += len(buf.Bytes())
-
-	return buf.Bytes(), nil
+	return nil
 }

mockserver/record.go

@@ -1,7 +1,7 @@
 package mockserver
 
 import (
-	"encoding/gob"
+	"encoding/binary"
 	"log"
 	"os"
 	"time"
@@ -9,6 +9,9 @@ import (
 	bngsdk "github.com/ESilva15/gobngsdk"
 )
 
+// NOTE: add some visual feedback of whats happening.
+// Maybe reuse the replay view function
+
 // Record records data from the UDP connection created by address and port
 func Record(address string, port int, filePath string) error {
 	ticker := time.NewTicker(time.Second / 60)
@@ -19,6 +22,7 @@ func Record(address string, port int, filePath string) error {
 	if err != nil {
 		return err
 	}
+	defer bin.Close()
 
 	// Create the BeamNGSDK instance
 	beam, err := bngsdk.Init(address, port)
@@ -27,15 +31,17 @@ func Record(address string, port int, filePath string) error {
 	}
 	defer beam.Close()
 
-	enc := gob.NewEncoder(bin)
 	for {
 		err := beam.ReadData()
 		if err != nil {
 			return err
 		}
-		if err := enc.Encode(beam.Data); err != nil {
+
+		err = binary.Write(bin, binary.LittleEndian, beam.Data)
+		if err != nil {
 			log.Fatal(err)
 		}
+
 		<-ticker.C
 	}
 }

mockserver/replay.go

@@ -3,50 +3,211 @@
 package mockserver
 
 import (
+	"bytes"
 	"context"
+	"encoding/binary"
 	"fmt"
 	"io"
 	"os"
+	"strings"
+	"sync"
 	"time"
+	"unsafe"
+
+	bngsdk "github.com/ESilva15/gobngsdk"
 )
 
-// Replay replays a given file <fp> in a UDP server <addr>:<port>
-func Replay(ctx context.Context, address string, port int, loop bool, fp string) error {
-	fileInfo, err := os.Stat(fp)
+type ViewData struct {
+	Data     []byte
+	SizeRead int64
+}
+
+// Replayer does the replaying
+// Should we make a "player" struct that can record and replay?
+type Replayer struct {
+	DataSourcePath string
+	Socket         *UDPTransport
+
+	// Streams
+	dataViewCh chan ViewData
+	socketCh   chan []byte
+
+	// Mut
+	mut  sync.RWMutex
+	data []byte
+
+	// View
+	viewData ViewData
+}
+
+func NewReplayer(address string, port int, fp string) (*Replayer, error) {
+	udp, err := NewUDPTransport(address, port)
 	if err != nil {
-		return fmt.Errorf("error stating file: %v", err)
+		return nil, err
 	}
 
-	bin, err := os.Open(fp)
+	replayer := &Replayer{
+		DataSourcePath: fp,
+		Socket:         udp,
+		data:           make([]byte, unsafe.Sizeof(bngsdk.Outgauge{})),
+		viewData:       ViewData{},
+		dataViewCh:     make(chan ViewData, 1),
+		socketCh:       make(chan []byte, 1),
+	}
+
+	return replayer, nil
+}
+
+// renderToTerminal will render the data for the users viewing pleasure
+func (r *Replayer) renderToTerminal(ctx context.Context) {
+	fileInfo, err := os.Stat(r.DataSourcePath)
+	if err != nil {
+		// NOTE: learn how to handle this error
+		// return fmt.Errorf("error stating file: %v", err)
+	}
+
+	// NOTE: temporary until I make a better view
+	var s strings.Builder
+	var bytesReader bytes.Reader
+	og := bngsdk.Outgauge{}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case data := <-r.dataViewCh:
+			// Reset to the start of the terminal
+			s.Reset()
+			fmt.Fprintf(os.Stdout, "\x1b[2J\x1b[H")
+
+			bytesReader.Reset(data.Data)
+			err := binary.Read(&bytesReader, binary.LittleEndian, &og)
+			if err != nil {
+				fmt.Fprintf(&s, "FAILED TO PARSE DATA\nError: %+v", err)
+				continue
+			}
+
+			percent := int(float64(data.SizeRead) / float64(fileInfo.Size()) * 100)
+			fmt.Fprintf(&s, "Replayed: %d%%\n", percent)
+			// NOTE: write a string serialization function on the SDK itself
+			fmt.Fprint(&s, "Outgauge {\n")
+			fmt.Fprintf(&s, "  Time:        %d ms\n", og.Time)
+			fmt.Fprintf(&s, "  Car:         %s\n", og.Car)
+			fmt.Fprintf(&s, "  Flags:       %b\n", og.Flags)
+			fmt.Fprintf(&s, "  Gear:        %d\n", og.Gear)
+			fmt.Fprintf(&s, "  Plid:        %d\n", og.Plid)
+			fmt.Fprintf(&s, "  Speed:       %f m/s\n", og.Speed)
+			fmt.Fprintf(&s, "  RPM:         %f RPM\n", og.RPM)
+			fmt.Fprintf(&s, "  Turbo:       %f Bar\n", og.Turbo)
+			fmt.Fprintf(&s, "  EngTemp:     %f °C\n", og.EngTemp)
+			fmt.Fprintf(&s, "  Fuel:        %f\n", og.Fuel)
+			fmt.Fprintf(&s, "  OilPressure: %f Bar\n", og.OilPressure)
+			fmt.Fprintf(&s, "  OilTemp:     %f °C\n", og.OilTemp)
+			fmt.Fprintf(&s, "  DashLights:  %b\n", og.DashLights)
+			fmt.Fprintf(&s, "  ShowLights:  %b\n", og.ShowLights)
+			fmt.Fprintf(&s, "  Throttle:    %f\n", og.Throttle)
+			fmt.Fprintf(&s, "  Brakes:      %f\n", og.Brake)
+			fmt.Fprintf(&s, "  Clutch:      %f\n", og.Clutch)
+			fmt.Fprintf(&s, "  Display1:    %s\n", og.Display1)
+			fmt.Fprintf(&s, "  Display2:    %s\n", og.Display2)
+			fmt.Fprintf(&s, "  ID:          %d\n", og.Display2)
+			fmt.Fprint(&s, "}\n\n")
+
+			fmt.Fprint(&s, "DashLights {\n")
+			fmt.Fprintf(&s, "  DL_SHIFT:      %t\n", og.DashLights&bngsdk.DL_SHIFT != 0)
+			fmt.Fprintf(&s, "  DL_FULLBEAM:   %t\n", og.DashLights&bngsdk.DL_FULLBEAM != 0)
+			fmt.Fprintf(&s, "  DL_HANDBRAKE:  %t\n", og.DashLights&bngsdk.DL_HANDBRAKE != 0)
+			fmt.Fprintf(&s, "  DL_PITSPEED:   %t\n", og.DashLights&bngsdk.DL_PITSPEED != 0)
+			fmt.Fprintf(&s, "  DL_TC:         %t\n", og.DashLights&bngsdk.DL_TC != 0)
+			fmt.Fprintf(&s, "  DL_SIGNAL_L:   %t\n", og.DashLights&bngsdk.DL_SIGNAL_L != 0)
+			fmt.Fprintf(&s, "  DL_SIGNAL_R:   %t\n", og.DashLights&bngsdk.DL_SIGNAL_R != 0)
+			fmt.Fprintf(&s, "  DL_SIGNAL_ANY: %t\n", og.DashLights&bngsdk.DL_SIGNAL_ANY != 0)
+			fmt.Fprintf(&s, "  DL_OILWARN:    %t\n", og.DashLights&bngsdk.DL_OILWARN != 0)
+			fmt.Fprintf(&s, "  DL_BATTERY:    %t\n", og.DashLights&bngsdk.DL_BATTERY != 0)
+			fmt.Fprintf(&s, "  DL_ABS:        %t\n", og.DashLights&bngsdk.DL_ABS != 0)
+			fmt.Fprintf(&s, "  DL_SPARE:      %t\n", og.DashLights&bngsdk.DL_SPARE != 0)
+			fmt.Fprint(&s, "}\n\n")
+
+			fmt.Fprint(&s, "ShowLights {\n") // Fixed typo "ShowLigths"
+			fmt.Fprintf(&s, "  DL_SHIFT:      %t\n", og.ShowLights&bngsdk.DL_SHIFT != 0)
+			fmt.Fprintf(&s, "  DL_FULLBEAM:   %t\n", og.ShowLights&bngsdk.DL_FULLBEAM != 0)
+			fmt.Fprintf(&s, "  DL_HANDBRAKE:  %t\n", og.ShowLights&bngsdk.DL_HANDBRAKE != 0)
+			fmt.Fprintf(&s, "  DL_PITSPEED:   %t\n", og.ShowLights&bngsdk.DL_PITSPEED != 0)
+			fmt.Fprintf(&s, "  DL_TC:         %t\n", og.ShowLights&bngsdk.DL_TC != 0)
+			fmt.Fprintf(&s, "  DL_SIGNAL_L:   %t\n", og.ShowLights&bngsdk.DL_SIGNAL_L != 0)
+			fmt.Fprintf(&s, "  DL_SIGNAL_R:   %t\n", og.ShowLights&bngsdk.DL_SIGNAL_R != 0)
+			fmt.Fprintf(&s, "  DL_SIGNAL_ANY: %t\n", og.ShowLights&bngsdk.DL_SIGNAL_ANY != 0)
+			fmt.Fprintf(&s, "  DL_OILWARN:    %t\n", og.ShowLights&bngsdk.DL_OILWARN != 0)
+			fmt.Fprintf(&s, "  DL_BATTERY:    %t\n", og.ShowLights&bngsdk.DL_BATTERY != 0)
+			fmt.Fprintf(&s, "  DL_ABS:        %t\n", og.ShowLights&bngsdk.DL_ABS != 0)
+			fmt.Fprintf(&s, "  DL_SPARE:      %t\n", og.ShowLights&bngsdk.DL_SPARE != 0)
+			fmt.Fprint(&s, "}\n\n")
+
+			fmt.Fprint(&s, "Flags {\n")
+			fmt.Fprintf(&s, "  OG_TURBO (Has Turbo): %t\n", og.Flags&bngsdk.OG_TURBO != 0)
+			fmt.Fprintf(&s, "  OG_KM (Is Metric):    %t\n", og.Flags&bngsdk.OG_KM != 0)
+			fmt.Fprintf(&s, "  OG_BAR (Pressure):    %t\n", og.Flags&bngsdk.OG_BAR != 0)
+			fmt.Fprint(&s, "}")
+
+			fmt.Fprint(os.Stdout, s.String())
+		}
+	}
+}
+
+// writeToUDPSocket will write the telemetry data to the UDP socket
+func (r *Replayer) writeToUDPSocket(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case data := <-r.socketCh:
+			r.mut.RLock()
+			_, err := r.Socket.Send(data)
+			r.mut.RUnlock()
+			if err != nil {
+				panic(fmt.Sprintf("error writing buffer to socket: %+v", err))
+				continue
+				// NOTE: log the error somewhere maybe
+				// return err
+			}
+		}
+	}
+}
+
+// Replay replays a given file <fp> in a UDP server <addr>:<port>
+func (r *Replayer) Replay(ctx context.Context, loop bool) error {
+	bin, err := os.Open(r.DataSourcePath)
 	if err != nil {
 		return fmt.Errorf("error opening file: %v", err)
 	}
 
 	reader := NewGobReader(bin)
-	udp, err := NewUDPTransport(address, port)
-	if err != nil {
-		return err
-	}
+
+	go r.renderToTerminal(ctx)
+	go r.writeToUDPSocket(ctx)
 
 	ticker := time.NewTicker(time.Second / 60)
 	defer ticker.Stop()
-
 	for {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
 
 		case <-ticker.C:
-			data, err := reader.Next()
+			r.mut.Lock()
+			err := reader.Next(r.data)
+			r.mut.Unlock()
+
 			if err == io.EOF {
 				if !loop {
-					return udp.Close()
+					return r.Socket.Close()
 				}
 
 				err = reader.Reset()
 				if err != nil {
 					return err
 				}
+
 				continue
 			}
 
@@ -54,13 +215,25 @@ func Replay(ctx context.Context, address string, port int, loop bool, fp string)
 				return err
 			}
 
-			_, err = udp.Send(data)
-			if err != nil {
-				return err
+			r.viewData.Data = r.data
+			r.viewData.SizeRead = reader.TotalRead
+
+			// Send the data to the view
+			select {
+			case r.dataViewCh <- r.viewData:
+				// Sent the data
+			default:
+				// Dropped the frame!
+			}
+
+			// Send the data to the UDP socket
+			select {
+			case r.socketCh <- r.data:
+				// Sent the data
+			default:
+				// Dropped the frame!
 			}
 
-			percent := int(float64(reader.TotalRead) / float64(fileInfo.Size()) * 100)
-			fmt.Printf("\rReplayed: %d%%", percent)
 		}
 	}
 }

mockserver/replay_test.go

@@ -0,0 +1,64 @@
+package mockserver
+
+import (
+	"context"
+	"net"
+	"os"
+	"testing"
+)
+
+func BenchmarkReplayAsync(b *testing.B) {
+	// Dummy socket
+	addr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
+	if err != nil {
+		b.Fatalf("failed to resolve UDP address: %v", err)
+	}
+
+	listener, err := net.ListenUDP("udp", addr)
+	if err != nil {
+		b.Fatalf("failed to start background UDP listener: %v", err)
+	}
+	defer listener.Close()
+
+	// Drain the socket
+	go func() {
+		buf := make([]byte, 65535)
+		for {
+			_, _, err := listener.ReadFrom(buf)
+			if err != nil {
+				return
+			}
+		}
+	}()
+
+	// Extract the random port assigned by the OS
+	assignedAddr := listener.LocalAddr().(*net.UDPAddr)
+
+	// Get the file path
+	filePath := "sunburstManual.bin"
+	if _, err := os.Stat(filePath); os.IsNotExist(err) {
+		filePath = "../" + filePath
+	}
+
+	// Reset the timer
+	b.ResetTimer()
+
+	// Benchmark loop
+	for i := 0; i < b.N; i++ {
+		ctx, cancel := context.WithCancel(context.Background())
+
+		replayer, err := NewReplayer(assignedAddr.IP.String(), assignedAddr.Port, filePath)
+		if err != nil {
+			cancel()
+			b.Fatalf("Error setting up replayer: %+v", err)
+		}
+
+		err = replayer.Replay(ctx, false)
+		if err != nil {
+			cancel()
+			b.Fatalf("Replay failed during benchmark run: %+v", err)
+		}
+
+		cancel()
+	}
+}