Fixed tests

- Fixed unit tests
- made pkg/stats thread-safe

internal/session/session.go

@@ -119,7 +119,8 @@ func (s *Session) createSessionDescription(desc webrtc.SessionDescription) error
 	if err != nil {
 		return err
 	}
-	fmt.Fprintf(s.sdpOutput, "Send this SDP:\n%s\n", resp)
+	fmt.Println("Send this SDP:")
+	fmt.Fprintf(s.sdpOutput, "%s\n", resp)
 	return nil
 }
 

pkg/session/bench/session_test.go

@@ -2,17 +2,21 @@ package bench
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
 
 func Test_OnNewDataChannel(t *testing.T) {
 	assert := assert.New(t)
+	testDuration := 2 * time.Second
 
 	sess := NewWith(Config{
 		Master: false,
 	})
 	assert.NotNil(sess)
+	sess.testDuration = testDuration
+	sess.testDurationError = (testDuration * 10) / 8
 
 	sess.onNewDataChannel()(nil)
 

pkg/session/bench/state_download.go

@@ -2,6 +2,7 @@ package bench
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/pion/webrtc/v2"
 	log "github.com/sirupsen/logrus"
@@ -30,9 +31,15 @@ func (s *Session) onOpenHandlerDownload(dc *webrtc.DataChannel) func() {
 			log.Warningln("No DataChannel provided")
 		}
 
+		timeoutErr := time.After(s.testDurationError)
 		fmt.Printf("Downloading random datas ... (%d s)\n", int(s.testDuration.Seconds()))
 
-		<-s.downloadDone
+		select {
+		case <-s.downloadDone:
+		case <-timeoutErr:
+			log.Error("Time'd out")
+		}
+
 		log.Traceln("Done downloading")
 
 		if !s.master {

pkg/session/session_test.go

@@ -69,6 +69,7 @@ func Test_TransferSmallMessage(t *testing.T) {
 	// Get SDP from sender and send it to the client
 	sdp, err := utils.MustReadStream(senderSDPOutput)
 	assert.Nil(err)
+	fmt.Printf("READ SDP -> %s\n", sdp)
 	sdp += "\n"
 	n, err = clientSDPProvider.WriteString(sdp)
 	assert.Nil(err)

pkg/stats/bytes.go

@@ -2,10 +2,16 @@ package stats
 
 // Bytes returns the stored number of bytes
 func (s *Stats) Bytes() uint64 {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return s.nbBytes
 }
 
 // AddBytes increase the nbBytes counter
 func (s *Stats) AddBytes(c uint64) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
 	s.nbBytes += c
 }

pkg/stats/ctrl.go

@@ -4,6 +4,9 @@ import "time"
 
 // Start stores the "start" timestamp
 func (s *Stats) Start() {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
 	if s.timeStart.IsZero() {
 		s.timeStart = time.Now()
 	} else if !s.timePause.IsZero() {
@@ -15,21 +18,37 @@ func (s *Stats) Start() {
 
 // Pause stores an interruption timestamp
 func (s *Stats) Pause() {
+	s.lock.RLock()
+
 	if s.timeStart.IsZero() || !s.timeStop.IsZero() {
 		// Can't stop if not started, or if stopped
+		s.lock.RUnlock()
 		return
 	}
+	s.lock.RUnlock()
+
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
 	if s.timePause.IsZero() {
 		s.timePause = time.Now()
 	}
 }
 
 // Stop stores the "stop" timestamp
 func (s *Stats) Stop() {
+	s.lock.RLock()
+
 	if s.timeStart.IsZero() {
 		// Can't stop if not started
+		s.lock.RUnlock()
 		return
 	}
+	s.lock.RUnlock()
+
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
 	if s.timeStop.IsZero() {
 		s.timeStop = time.Now()
 	}

pkg/stats/data.go

@@ -6,6 +6,9 @@ import "time"
 // Returns 0 if not started
 // Returns time.Since(s.timeStart) if not stopped
 func (s *Stats) Duration() time.Duration {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	if s.timeStart.IsZero() {
 		return 0
 	} else if s.timeStop.IsZero() {
@@ -16,5 +19,8 @@ func (s *Stats) Duration() time.Duration {
 
 // Bandwidth returns the IO speed in MB/s
 func (s *Stats) Bandwidth() float64 {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
 	return (float64(s.nbBytes) / 1024 / 1024) / s.Duration().Seconds()
 }

pkg/stats/stats.go

@@ -2,11 +2,13 @@ package stats
 
 import (
 	"fmt"
+	"sync"
 	"time"
 )
 
 // Stats provide a way to track statistics infos
 type Stats struct {
+	lock      sync.RWMutex
 	nbBytes   uint64
 	timeStart time.Time
 	timeStop  time.Time
@@ -16,5 +18,7 @@ type Stats struct {
 }
 
 func (s *Stats) String() string {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
 	return fmt.Sprintf("%v bytes | %-v | %0.4f MB/s", s.Bytes(), s.Duration(), s.Bandwidth())
 }