Repository: bluenviron/gortsplib
Pull request #681 (2025-01-17 16:49:09)

Fix writer nullpointer panic on network reconnect
In https://github.com/bluenviron/gortsplib/pull/655 the change was made in `client.go` to set `writer` to `nil`, before completing the 'closing' of the client. This makes that if someone calls `WritePacketRTPWithNTP` after `writer = nil` but before `c.done` is closed, there is a panic. As described in https://github.com/bluenviron/gortsplib/issues/678. 

I don't fully understand why `c.writer = nil` is needed in the first place. It seems that before https://github.com/bluenviron/gortsplib/pull/655 `c.writer` is never set to `nil`. And after testing the situation of the panic, this resolved it. If someone has a better solution I'm very eager to learn!

aler9: `writer` was set to `nil` to prevent `chWriterError` from returning a closed channel and therefore messing up the client loop. Anyway, this was causing issues both when the client was closing and when the client was paused, so the fix is right. Thanks.

Patch:
diff --git a/async_processor.go b/async_processor.go
index 660b03bc..2a0a9f0a 100644
--- a/async_processor.go
+++ b/async_processor.go
@@ -10,10 +10,11 @@ import (
 type asyncProcessor struct {
 	bufferSize int
 
-	running bool
-	buffer  *ringbuffer.RingBuffer
+	running   bool
+	buffer    *ringbuffer.RingBuffer
+	stopError error
 
-	chError chan error
+	stopped chan struct{}
 }
 
 func (w *asyncProcessor) initialize() {
@@ -22,22 +23,21 @@ func (w *asyncProcessor) initialize() {
 
 func (w *asyncProcessor) start() {
 	w.running = true
-	w.chError = make(chan error)
+	w.stopped = make(chan struct{})
 	go w.run()
 }
 
 func (w *asyncProcessor) stop() {
 	if w.running {
 		w.buffer.Close()
-		<-w.chError
+		<-w.stopped
 		w.running = false
 	}
 }
 
 func (w *asyncProcessor) run() {
-	err := w.runInner()
-	w.chError <- err
-	close(w.chError)
+	w.stopError = w.runInner()
+	close(w.stopped)
 }
 
 func (w *asyncProcessor) runInner() error {
diff --git a/client.go b/client.go
index 5dfaad01..be014ed7 100644
--- a/client.go
+++ b/client.go
@@ -559,9 +559,9 @@ func (c *Client) runInner() error {
 			return nil
 		}()
 
-		chWriterError := func() chan error {
-			if c.writer != nil {
-				return c.writer.chError
+		chWriterError := func() chan struct{} {
+			if c.writer != nil && c.writer.running {
+				return c.writer.stopped
 			}
 			return nil
 		}()
@@ -637,8 +637,8 @@ func (c *Client) runInner() error {
 			}
 			c.keepaliveTimer = time.NewTimer(c.keepalivePeriod)
 
-		case err := <-chWriterError:
-			return err
+		case <-chWriterError:
+			return c.writer.stopError
 
 		case err := <-chReaderError:
 			c.reader = nil
@@ -911,8 +911,6 @@ func (c *Client) stopTransportRoutines() {
 	}
 
 	c.timeDecoder = nil
-
-	c.writer = nil
 }
 
 func (c *Client) connOpen() error {
diff --git a/server_session.go b/server_session.go
index 549f388b..1c3c77bc 100644
--- a/server_session.go
+++ b/server_session.go
@@ -626,9 +626,9 @@ func (ss *ServerSession) run() {
 
 func (ss *ServerSession) runInner() error {
 	for {
-		chWriterError := func() chan error {
-			if ss.writer != nil {
-				return ss.writer.chError
+		chWriterError := func() chan struct{} {
+			if ss.writer != nil && ss.writer.running {
+				return ss.writer.stopped
 			}
 			return nil
 		}()
@@ -729,8 +729,8 @@ func (ss *ServerSession) runInner() error {
 
 			ss.udpCheckStreamTimer = time.NewTimer(ss.s.checkStreamPeriod)
 
-		case err := <-chWriterError:
-			return err
+		case <-chWriterError:
+			return ss.writer.stopError
 
 		case <-ss.ctx.Done():
 			return liberrors.ErrServerTerminated{}
@@ -1306,7 +1306,6 @@ func (ss *ServerSession) handleRequestInner(sc *ServerConn, req *base.Request) (
 			}
 
 			ss.writer.stop()
-			ss.writer = nil
 
 			ss.timeDecoder = nil
 


Test patch:
diff --git a/async_processor_test.go b/async_processor_test.go
new file mode 100644
index 00000000..2f881bd6
--- /dev/null
+++ b/async_processor_test.go
@@ -0,0 +1,24 @@
+package gortsplib
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAsyncProcessorStopAfterError(t *testing.T) {
+	p := &asyncProcessor{bufferSize: 8}
+	p.initialize()
+
+	p.push(func() error {
+		return fmt.Errorf("ok")
+	})
+
+	p.start()
+
+	<-p.stopped
+	require.EqualError(t, p.stopError, "ok")
+
+	p.stop()
+}
diff --git a/client_play_test.go b/client_play_test.go
index 87f86cbf..05a6dc61 100644
--- a/client_play_test.go
+++ b/client_play_test.go
@@ -1961,8 +1961,8 @@ func TestClientPlayPause(t *testing.T) {
 				})
 				require.NoError(t, err2)
 
-				req, err = conn.ReadRequest()
-				require.NoError(t, err)
+				req, err2 = conn.ReadRequest()
+				require.NoError(t, err2)
 				require.Equal(t, base.Play, req.Method)
 
 				err2 = conn.WriteResponse(&base.Response{