Merge pull request #6 from samespace/mute

Add force mute/unmute of client

client.go

@@ -147,6 +147,7 @@ type Client struct {
 	mu                    sync.RWMutex
 	peerConnection        *PeerConnection
 	quicClient            quic.Connection
+	isMuted               *atomic.Bool
 	// pending received tracks are the remote tracks from other clients that waiting to add when the client is connected
 	pendingReceivedTracks []SubscribeTrackRequest
 	// pending published tracks are the remote tracks that still state as unknown source, and can't be published until the client state the source media or screen
@@ -329,6 +330,7 @@ func NewClient(s *SFU, id string, name string, peerConnectionConfig webrtc.Confi
 		mu:                             sync.RWMutex{},
 		negotiationNeeded:              &atomic.Bool{},
 		peerConnection:                 newPeerConnection(peerConnection),
+		isMuted:                        &atomic.Bool{},
 		state:                          &stateNew,
 		tracks:                         newTrackList(opts.Log),
 		options:                        opts,
@@ -585,6 +587,24 @@ func NewClient(s *SFU, id string, name string, peerConnectionConfig webrtc.Confi
 	return client
 }
 
+func (c *Client) IsMuted() bool {
+	return c.isMuted.Load()
+}
+
+func (c *Client) Mute() {
+	c.isMuted.CompareAndSwap(false, true)
+	for _, track := range c.tracks.GetTracks() {
+		track.Mute()
+	}
+}
+
+func (c *Client) Unmute() {
+	c.isMuted.CompareAndSwap(true, false)
+	for _, track := range c.tracks.GetTracks() {
+		track.Unmute()
+	}
+}
+
 /*
 Starts an indivisual client recording session.
 Creates a new quic client solely for this client.

examples/http-websocket/main.go

@@ -162,6 +162,7 @@ func main() {
 	roomsOpts.Bitrates.InitialBandwidth = 1_000_000
 	// roomsOpts.PLIInterval = 3 * time.Second
 	defaultRoom, _ := roomManager.NewRoom(roomID, roomName, sfu.RoomTypeLocal, roomsOpts)
+
 	// turnServer := sfu.StartTurnServer(ctx, localIp.String())
 	// defer turnServer.Close()
 

track.go

@@ -70,6 +70,8 @@ type ITrack interface {
 	StopRecording()
 	PauseRecording()
 	ContinueRecording()
+	Mute()
+	Unmute()
 }
 
 type Track struct {
@@ -82,6 +84,7 @@ type Track struct {
 	trackRecorder    recorder.TrackRecorder
 	isRecording      atomic.Bool
 	isPaused         atomic.Bool
+	isMuted          atomic.Bool
 }
 
 func newTrack(ctx context.Context, client *Client, trackRemote IRemoteTrack, minWait, maxWait, pliInterval time.Duration, onPLI func(), stats stats.Getter, onStatsUpdated func(*stats.Stats)) (ITrack, error) {
@@ -106,10 +109,16 @@ func newTrack(ctx context.Context, client *Client, trackRemote IRemoteTrack, min
 		onEndedCallbacks: make([]func(), 0),
 		isRecording:      atomic.Bool{},
 		isPaused:         atomic.Bool{},
+		isMuted:          atomic.Bool{},
 	}
 
 	onRead := func(p *rtp.Packet) {
 		tracks := t.base.clientTracks.GetTracks()
+		if t.MimeType() == webrtc.MimeTypeOpus {
+			if t.isMuted.Load() {
+				p = t.getSilencePacket(p)
+			}
+		}
 
 		for _, track := range tracks {
 			//nolint:ineffassign,staticcheck // packet is from the pool
@@ -171,6 +180,14 @@ func newTrack(ctx context.Context, client *Client, trackRemote IRemoteTrack, min
 	return t, nil
 }
 
+func (t *Track) Mute() {
+	t.isMuted.CompareAndSwap(false, true)
+}
+
+func (t *Track) Unmute() {
+	t.isMuted.CompareAndSwap(true, false)
+}
+
 func (t *Track) StartRecording(stream quic.SendStream) error {
 	t.mu.Lock()
 	defer t.mu.Unlock()
@@ -499,6 +516,9 @@ func newSimulcastTrack(client *Client, track IRemoteTrack, minWait, maxWait, pli
 	return t
 }
 
+func (t *SimulcastTrack) Mute()   {}
+func (t *SimulcastTrack) Unmute() {}
+
 func (t *SimulcastTrack) StartRecording(stream quic.SendStream) error {
 	return nil
 }
@@ -621,6 +641,7 @@ func (t *SimulcastTrack) AddRemoteTrack(track IRemoteTrack, minWait, maxWait tim
 		}
 
 		tracks := t.base.clientTracks.GetTracks()
+
 		for _, track := range tracks {
 			//nolint:ineffassign,staticcheck // packet is from the pool
 			packet := t.base.pool.NewPacket(&p.Header, p.Payload)