client: Log bandwidth used (incoming / outgoing).

client/main.go

@@ -73,40 +73,6 @@ const (
 	maxMessageSize = 64 * 1024
 )
 
-type Stats struct {
-	numRecvMessages   atomic.Int64
-	numSentMessages   atomic.Int64
-	resetRecvMessages int64
-	resetSentMessages int64
-
-	start time.Time
-}
-
-func (s *Stats) reset(start time.Time) {
-	s.resetRecvMessages = s.numRecvMessages.Load()
-	s.resetSentMessages = s.numSentMessages.Load()
-	s.start = start
-}
-
-func (s *Stats) Log() {
-	now := time.Now()
-	duration := now.Sub(s.start)
-	perSec := int64(duration / time.Second)
-	if perSec == 0 {
-		return
-	}
-
-	totalSentMessages := s.numSentMessages.Load()
-	sentMessages := totalSentMessages - s.resetSentMessages
-	totalRecvMessages := s.numRecvMessages.Load()
-	recvMessages := totalRecvMessages - s.resetRecvMessages
-	log.Printf("Stats: sent=%d (%d/sec), recv=%d (%d/sec), delta=%d",
-		totalSentMessages, sentMessages/perSec,
-		totalRecvMessages, recvMessages/perSec,
-		totalSentMessages-totalRecvMessages)
-	s.reset(now)
-}
-
 type MessagePayload struct {
 	Now time.Time `json:"now"`
 }
@@ -283,6 +249,8 @@ func (c *SignalingClient) readPump() {
 			break
 		}
 
+		c.stats.numRecvBytes.Add(uint64(decodeBuffer.Len()))
+
 		var message signaling.ServerMessage
 		if err := message.UnmarshalJSON(decodeBuffer.Bytes()); err != nil {
 			log.Printf("Error: %v", err)
@@ -297,9 +265,10 @@ func (c *SignalingClient) writeInternal(message *signaling.ClientMessage) bool {
 	var closeData []byte
 
 	c.conn.SetWriteDeadline(time.Now().Add(writeWait)) // nolint
+	var written int
 	writer, err := c.conn.NextWriter(websocket.TextMessage)
 	if err == nil {
-		_, err = easyjson.MarshalToWriter(message, writer)
+		written, err = easyjson.MarshalToWriter(message, writer)
 	}
 	if err != nil {
 		if err == websocket.ErrCloseSent {
@@ -315,6 +284,9 @@ func (c *SignalingClient) writeInternal(message *signaling.ClientMessage) bool {
 
 	writer.Close()
 	c.stats.numSentMessages.Add(1)
+	if written > 0 {
+		c.stats.numSentBytes.Add(uint64(written))
+	}
 	return true
 
 close:

client/stats.go

@@ -0,0 +1,97 @@
+/**
+ * Standalone signaling server for the Nextcloud Spreed app.
+ * Copyright (C) 2025 struktur AG
+ *
+ * @author Joachim Bauch <bauch@struktur.de>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package main
+
+import (
+	"log"
+	"sync/atomic"
+	"time"
+
+	"github.com/strukturag/nextcloud-spreed-signaling/api"
+)
+
+type Stats struct {
+	numRecvMessages   atomic.Int64
+	numSentMessages   atomic.Int64
+	resetRecvMessages int64
+	resetSentMessages int64
+	numRecvBytes      atomic.Uint64
+	numSentBytes      atomic.Uint64
+	resetRecvBytes    uint64
+	resetSentBytes    uint64
+
+	start time.Time
+}
+
+func (s *Stats) reset(start time.Time) {
+	s.resetRecvMessages = s.numRecvMessages.Load()
+	s.resetSentMessages = s.numSentMessages.Load()
+	s.resetRecvBytes = s.numRecvBytes.Load()
+	s.resetSentBytes = s.numSentBytes.Load()
+	s.start = start
+}
+
+type statsLogEntries struct {
+	totalSentMessages  int64
+	sentMessagesPerSec int64
+	sentBytesPerSec    api.Bandwidth
+
+	totalRecvMessages  int64
+	recvMessagesPerSec int64
+	recvBytesPerSec    api.Bandwidth
+}
+
+func (s *Stats) getLogEntries(now time.Time) *statsLogEntries {
+	duration := now.Sub(s.start)
+	perSec := int64(duration / time.Second)
+	if perSec == 0 {
+		return nil
+	}
+
+	totalSentMessages := s.numSentMessages.Load()
+	sentMessages := totalSentMessages - s.resetSentMessages
+	sentBytes := api.BandwidthFromBytes(s.numSentBytes.Load() - s.resetSentBytes)
+	totalRecvMessages := s.numRecvMessages.Load()
+	recvMessages := totalRecvMessages - s.resetRecvMessages
+	recvBytes := api.BandwidthFromBytes(s.numRecvBytes.Load() - s.resetRecvBytes)
+
+	s.reset(now)
+	return &statsLogEntries{
+		totalSentMessages:  totalSentMessages,
+		sentMessagesPerSec: sentMessages / perSec,
+		sentBytesPerSec:    sentBytes,
+
+		totalRecvMessages:  totalRecvMessages,
+		recvMessagesPerSec: recvMessages / perSec,
+		recvBytesPerSec:    recvBytes,
+	}
+}
+
+func (s *Stats) Log() {
+	now := time.Now()
+	if entries := s.getLogEntries(now); entries != nil {
+		log.Printf("Stats: sent=%d (%d/sec, %s), recv=%d (%d/sec, %s), delta=%d",
+			entries.totalSentMessages, entries.sentMessagesPerSec, entries.sentBytesPerSec,
+			entries.totalRecvMessages, entries.recvMessagesPerSec, entries.recvBytesPerSec,
+			entries.totalSentMessages-entries.totalRecvMessages)
+	}
+}

client/stats_test.go

@@ -0,0 +1,79 @@
+/**
+ * Standalone signaling server for the Nextcloud Spreed app.
+ * Copyright (C) 2025 struktur AG
+ *
+ * @author Joachim Bauch <bauch@struktur.de>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package main
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/strukturag/nextcloud-spreed-signaling/api"
+)
+
+func TestStats(t *testing.T) {
+	assert := assert.New(t)
+
+	var stats Stats
+	assert.Nil(stats.getLogEntries(time.Time{}))
+	now := time.Now()
+	if entries := stats.getLogEntries(now); assert.NotNil(entries) {
+		assert.EqualValues(0, entries.totalSentMessages)
+		assert.EqualValues(0, entries.sentMessagesPerSec)
+		assert.EqualValues(0, entries.sentBytesPerSec)
+
+		assert.EqualValues(0, entries.totalRecvMessages)
+		assert.EqualValues(0, entries.recvMessagesPerSec)
+		assert.EqualValues(0, entries.recvBytesPerSec)
+	}
+
+	stats.numSentMessages.Add(10)
+	stats.numSentBytes.Add((api.Bandwidth(20) * api.Kilobit).Bits())
+
+	stats.numRecvMessages.Add(30)
+	stats.numRecvBytes.Add((api.Bandwidth(40) * api.Kilobit).Bits())
+
+	if entries := stats.getLogEntries(now.Add(time.Second)); assert.NotNil(entries) {
+		assert.EqualValues(10, entries.totalSentMessages)
+		assert.EqualValues(10, entries.sentMessagesPerSec)
+		assert.EqualValues(20*1024*8, entries.sentBytesPerSec)
+
+		assert.EqualValues(30, entries.totalRecvMessages)
+		assert.EqualValues(30, entries.recvMessagesPerSec)
+		assert.EqualValues(40*1024*8, entries.recvBytesPerSec)
+	}
+
+	stats.numSentMessages.Add(100)
+	stats.numSentBytes.Add((api.Bandwidth(200) * api.Kilobit).Bits())
+
+	stats.numRecvMessages.Add(300)
+	stats.numRecvBytes.Add((api.Bandwidth(400) * api.Kilobit).Bits())
+
+	if entries := stats.getLogEntries(now.Add(2 * time.Second)); assert.NotNil(entries) {
+		assert.EqualValues(110, entries.totalSentMessages)
+		assert.EqualValues(100, entries.sentMessagesPerSec)
+		assert.EqualValues(200*1024*8, entries.sentBytesPerSec)
+
+		assert.EqualValues(330, entries.totalRecvMessages)
+		assert.EqualValues(300, entries.recvMessagesPerSec)
+		assert.EqualValues(400*1024*8, entries.recvBytesPerSec)
+	}
+}