Repository: bluenviron/gortsplib

Issue:
nil pointer dereference panic after network reconnect
I'm seeing a nullpointer dereference panic at this file at line `1928`:

https://github.com/bluenviron/gortsplib/blob/c308cb3355c01e2d02230dc2ae8705b39c201b8f/client.go#L1909-L1936

It happens when the connection is harshly broken as a result of a network disruption.

I suspect there is some race condition caused by this line:

https://github.com/bluenviron/gortsplib/blob/c308cb3355c01e2d02230dc2ae8705b39c201b8f/client.go#L915

The trace is:
```
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x10 pc=0xb4a308]

goroutine 14 [running]:
github.com/bluenviron/gortsplib/v4.(*asyncProcessor).push(...)
	/home/ascento/go/pkg/mod/github.com/bluenviron/gortsplib/v4@v4.12.1/async_processor.go:58
github.com/bluenviron/gortsplib/v4.(*Client).WritePacketRTPWithNTP(0x4000544fc8, 0x400054e0a0, 0x40007e07a0, {0x5f3a38?, 0x40008ebaa8?, 0x15253c0?})
	/home/ascento/go/pkg/mod/github.com/bluenviron/gortsplib/v4@v4.12.1/client.go:1928 +0x178
```

Discussion:
@SijmenHuizenga: I've been trying to pinpoint the cause, and my current best guess is that this is introduced in #655 

Unrelated: The work in #655 is a huge step forward of working in poor network connectivity areas. Looking forward to being able to roll it out after fixing this issue.@aler9: Fixed by #655@github-actions: This issue is being locked automatically because it has been closed for more than 6 months.
Please open a new issue in case you encounter a similar problem.

Solution 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{