Merge pull request #209 from screego/stuff

go.mod

@@ -13,20 +13,20 @@ require (
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/pion/randutil v0.1.0
 	github.com/pion/turn/v4 v4.0.0
-	github.com/prometheus/client_golang v1.20.4
+	github.com/prometheus/client_golang v1.20.5
 	github.com/rs/xid v1.6.0
 	github.com/rs/zerolog v1.33.0
-	github.com/stretchr/testify v1.9.0
-	github.com/urfave/cli v1.22.15
-	golang.org/x/crypto v0.27.0
-	golang.org/x/term v0.24.0
-	golang.org/x/text v0.18.0
+	github.com/stretchr/testify v1.10.0
+	github.com/urfave/cli v1.22.16
+	golang.org/x/crypto v0.30.0
+	golang.org/x/term v0.27.0
+	golang.org/x/text v0.21.0
 )
 
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/cpuguy83/go-md2man/v2 v2.0.4 // indirect
+	github.com/cpuguy83/go-md2man/v2 v2.0.5 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/felixge/httpsnoop v1.0.3 // indirect
 	github.com/gorilla/securecookie v1.1.2 // indirect
@@ -45,7 +45,7 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/wlynxg/anet v0.0.4 // indirect
-	golang.org/x/sys v0.25.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
 	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,11 +1,14 @@
 github.com/BurntSushi/toml v1.3.2/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
+github.com/BurntSushi/toml v1.4.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/cpuguy83/go-md2man/v2 v2.0.4 h1:wfIWP927BUkWJb2NmU/kNDYIBTh/ziUX91+lVfRxZq4=
 github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/cpuguy83/go-md2man/v2 v2.0.5 h1:ZtcqGrnekaHpVLArFSe4HK5DoKx1T0rq2DwVB0alcyc=
+github.com/cpuguy83/go-md2man/v2 v2.0.5/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -63,6 +66,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v1.20.4 h1:Tgh3Yr67PaOv/uTqloMsCEdeuFTatm5zIq5+qNN23vI=
 github.com/prometheus/client_golang v1.20.4/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
+github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y=
+github.com/prometheus/client_golang v1.20.5/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
 github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
 github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
 github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G1dc=
@@ -87,23 +92,35 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/urfave/cli v1.22.15 h1:nuqt+pdC/KqswQKhETJjo7pvn/k4xMUxgW6liI7XpnM=
 github.com/urfave/cli v1.22.15/go.mod h1:wSan1hmo5zeyLGBjRJbzRTNk8gwoYa2B9n4q9dmRIc0=
+github.com/urfave/cli v1.22.16 h1:MH0k6uJxdwdeWQTwhSO42Pwr4YLrNLwBtg1MRgTqPdQ=
+github.com/urfave/cli v1.22.16/go.mod h1:EeJR6BKodywf4zciqrdw6hpCPk68JO9z5LazXZMn5Po=
 github.com/wlynxg/anet v0.0.4 h1:0de1OFQxnNqAu+x2FAKKCVIrnfGKQbs7FQz++tB0+Uw=
 github.com/wlynxg/anet v0.0.4/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
 golang.org/x/crypto v0.27.0 h1:GXm2NjJrPaiv/h1tb2UH8QfgC/hOf/+z0p6PT8o1w7A=
 golang.org/x/crypto v0.27.0/go.mod h1:1Xngt8kV6Dvbssa53Ziq6Eqn0HqbZi5Z6R0ZpwQzt70=
+golang.org/x/crypto v0.30.0 h1:RwoQn3GkWiMkzlX562cLB7OxWvjH1L8xutO2WoJcRoY=
+golang.org/x/crypto v0.30.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/net v0.27.0 h1:5K3Njcw06/l2y9vpGCSdcxWOYHOUk3dVNGDXN+FvAys=
 golang.org/x/net v0.27.0/go.mod h1:dDi0PyhWNoiUOrAS8uXv/vnScO4wnHQO4mj9fn/RytE=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
 golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.24.0 h1:Mh5cbb+Zk2hqqXNO7S1iTjEphVL+jb8ZWaqh/g+JWkM=
 golang.org/x/term v0.24.0/go.mod h1:lOBK/LVxemqiMij05LGJ0tzNr8xlmwBRJ81PX6wVLH8=
+golang.org/x/term v0.27.0 h1:WP60Sv1nlK1T6SupCHbXzSaN0b9wUmsPoRS9b61A23Q=
+golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
 golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
 golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
 google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

ws/event_clientanswer.go

@@ -32,7 +32,7 @@ func (e *ClientAnswer) Execute(rooms *Rooms, current ClientInfo) error {
 		return fmt.Errorf("permission denied for session %s", e.SID)
 	}
 
-	room.Users[session.Host].Write <- outgoing.ClientAnswer(*e)
+	room.Users[session.Host].WriteTimeout(outgoing.ClientAnswer(*e))
 
 	return nil
 }

ws/event_clientice.go

@@ -32,7 +32,7 @@ func (e *ClientICE) Execute(rooms *Rooms, current ClientInfo) error {
 		return fmt.Errorf("permission denied for session %s", e.SID)
 	}
 
-	room.Users[session.Host].Write <- outgoing.ClientICE(*e)
+	room.Users[session.Host].WriteTimeout(outgoing.ClientICE(*e))
 
 	return nil
 }

ws/event_create.go

@@ -70,7 +70,7 @@ func (e *Create) Execute(rooms *Rooms, current ClientInfo) error {
 				Streaming: false,
 				Owner:     true,
 				Addr:      current.Addr,
-				Write:     current.Write,
+				_write:    current.Write,
 			},
 		},
 	}

ws/event_disconnected.go

@@ -20,7 +20,7 @@ func (e *Disconnected) Execute(rooms *Rooms, current ClientInfo) error {
 func (e *Disconnected) executeNoError(rooms *Rooms, current ClientInfo) {
 	roomID := rooms.connected[current.ID]
 	delete(rooms.connected, current.ID)
-	current.Write <- outgoing.CloseWriter{Code: e.Code, Reason: e.Reason}
+	writeTimeout[outgoing.Message](current.Write, outgoing.CloseWriter{Code: e.Code, Reason: e.Reason})
 
 	if roomID == "" {
 		return
@@ -46,14 +46,14 @@ func (e *Disconnected) executeNoError(rooms *Rooms, current ClientInfo) {
 		if bytes.Equal(session.Client.Bytes(), current.ID.Bytes()) {
 			host, ok := room.Users[session.Host]
 			if ok {
-				host.Write <- outgoing.EndShare(id)
+				host.WriteTimeout(outgoing.EndShare(id))
 			}
 			room.closeSession(rooms, id)
 		}
 		if bytes.Equal(session.Host.Bytes(), current.ID.Bytes()) {
 			client, ok := room.Users[session.Client]
 			if ok {
-				client.Write <- outgoing.EndShare(id)
+				client.WriteTimeout(outgoing.EndShare(id))
 			}
 			room.closeSession(rooms, id)
 		}
@@ -62,7 +62,7 @@ func (e *Disconnected) executeNoError(rooms *Rooms, current ClientInfo) {
 	if user.Owner && room.CloseOnOwnerLeave {
 		for _, member := range room.Users {
 			delete(rooms.connected, member.ID)
-			member.Write <- outgoing.CloseWriter{Code: websocket.CloseNormalClosure, Reason: CloseOwnerLeft}
+			member.WriteTimeout(outgoing.CloseWriter{Code: websocket.CloseNormalClosure, Reason: CloseOwnerLeft})
 		}
 		rooms.closeRoom(roomID)
 		return

ws/event_health.go

@@ -5,6 +5,6 @@ type Health struct {
 }
 
 func (e *Health) Execute(rooms *Rooms, current ClientInfo) error {
-	e.Response <- len(rooms.connected)
+	writeTimeout(e.Response, len(rooms.connected))
 	return nil
 }

ws/event_hostice.go

@@ -32,7 +32,7 @@ func (e *HostICE) Execute(rooms *Rooms, current ClientInfo) error {
 		return fmt.Errorf("permission denied for session %s", e.SID)
 	}
 
-	room.Users[session.Client].Write <- outgoing.HostICE(*e)
+	room.Users[session.Client].WriteTimeout(outgoing.HostICE(*e))
 
 	return nil
 }

ws/event_hostoffer.go

@@ -32,7 +32,7 @@ func (e *HostOffer) Execute(rooms *Rooms, current ClientInfo) error {
 		return fmt.Errorf("permission denied for session %s", e.SID)
 	}
 
-	room.Users[session.Client].Write <- outgoing.HostOffer(*e)
+	room.Users[session.Client].WriteTimeout(outgoing.HostOffer(*e))
 
 	return nil
 }

ws/event_join.go

@@ -38,7 +38,7 @@ func (e *Join) Execute(rooms *Rooms, current ClientInfo) error {
 		Streaming: false,
 		Owner:     false,
 		Addr:      current.Addr,
-		Write:     current.Write,
+		_write:    current.Write,
 	}
 	rooms.connected[current.ID] = room.ID
 	room.notifyInfoChanged()

ws/event_stop_share.go

@@ -25,7 +25,7 @@ func (e *StopShare) Execute(rooms *Rooms, current ClientInfo) error {
 		if bytes.Equal(session.Host.Bytes(), current.ID.Bytes()) {
 			client, ok := room.Users[session.Client]
 			if ok {
-				client.Write <- outgoing.EndShare(id)
+				client.WriteTimeout(outgoing.EndShare(id))
 			}
 			room.closeSession(rooms, id)
 		}

ws/room.go

@@ -4,8 +4,10 @@ import (
 	"fmt"
 	"net"
 	"sort"
+	"time"
 
 	"github.com/rs/xid"
+	"github.com/rs/zerolog/log"
 	"github.com/screego/server/config"
 	"github.com/screego/server/ws/outgoing"
 )
@@ -60,8 +62,8 @@ func (r *Room) newSession(host, client xid.ID, rooms *Rooms, v4, v6 net.IP) {
 			Username:   clientName,
 		}}
 	}
-	r.Users[host].Write <- outgoing.HostSession{Peer: client, ID: id, ICEServers: iceHost}
-	r.Users[client].Write <- outgoing.ClientSession{Peer: host, ID: id, ICEServers: iceClient}
+	r.Users[host].WriteTimeout(outgoing.HostSession{Peer: client, ID: id, ICEServers: iceHost})
+	r.Users[client].WriteTimeout(outgoing.ClientSession{Peer: host, ID: id, ICEServers: iceClient})
 }
 
 func (r *Rooms) addresses(prefix string, v4, v6 net.IP, tcp bool) (result []string) {
@@ -122,10 +124,10 @@ func (r *Room) notifyInfoChanged() {
 			return left.Name < right.Name
 		})
 
-		current.Write <- outgoing.Room{
+		current.WriteTimeout(outgoing.Room{
 			ID:    r.ID,
 			Users: users,
-		}
+		})
 	}
 }
 
@@ -135,5 +137,17 @@ type User struct {
 	Name      string
 	Streaming bool
 	Owner     bool
-	Write     chan<- outgoing.Message
+	_write    chan<- outgoing.Message
+}
+
+func (u *User) WriteTimeout(msg outgoing.Message) {
+	writeTimeout(u._write, msg)
+}
+
+func writeTimeout[T any](ch chan<- T, msg T) {
+	select {
+	case <-time.After(2 * time.Second):
+		log.Warn().Interface("event", fmt.Sprintf("%T", msg)).Interface("payload", msg).Msg("Client write loop didn't accept the message.")
+	case ch <- msg:
+	}
 }

ws/rooms.go

@@ -111,17 +111,18 @@ func (r *Rooms) Start() {
 }
 
 func (r *Rooms) Count() (int, string) {
+	timeout := time.After(5 * time.Second)
+
 	h := Health{Response: make(chan int, 1)}
 	select {
 	case r.Incoming <- ClientMessage{SkipConnectedCheck: true, Incoming: &h}:
-	case <-time.After(5 * time.Second):
+	case <-timeout:
 		return -1, "main loop didn't accept a message within 5 second"
 	}
-	r.Incoming <- ClientMessage{SkipConnectedCheck: true, Incoming: &h}
 	select {
 	case count := <-h.Response:
 		return count, ""
-	case <-time.After(5 * time.Second):
+	case <-timeout:
 		return -1, "main loop didn't respond to a message within 5 second"
 	}
 }