client: remove channel from ReadFrames()

client_publish_test.go

@@ -181,12 +181,16 @@ func TestClientPublishSerial(t *testing.T) {
 			require.NoError(t, err)
 
 			recvDone := make(chan struct{})
-			done := conn.ReadFrames(func(trackID int, streamType StreamType, payload []byte) {
+			done := make(chan struct{})
-				require.Equal(t, 0, trackID)
+			go func() {
-				require.Equal(t, StreamTypeRTCP, streamType)
+				defer close(done)
-				require.Equal(t, []byte{0x05, 0x06, 0x07, 0x08}, payload)
+				conn.ReadFrames(func(trackID int, streamType StreamType, payload []byte) {
-				close(recvDone)
+					require.Equal(t, 0, trackID)
-			})
+					require.Equal(t, StreamTypeRTCP, streamType)
+					require.Equal(t, []byte{0x05, 0x06, 0x07, 0x08}, payload)
+					close(recvDone)
+				})
+			}()
 
 			err = conn.WriteFrame(track.ID, StreamTypeRTP,
 				[]byte{0x01, 0x02, 0x03, 0x04})

client_read_test.go

@@ -359,20 +359,24 @@ func TestClientRead(t *testing.T) {
 			conn, err := c.DialRead(scheme + "://localhost:8554/teststream")
 			require.NoError(t, err)
 
-			done := conn.ReadFrames(func(id int, streamType StreamType, payload []byte) {
+			done := make(chan struct{})
-				require.Equal(t, 0, id)
+			go func() {
-				require.Equal(t, StreamTypeRTP, streamType)
+				defer close(done)
-				require.Equal(t, []byte{0x01, 0x02, 0x03, 0x04}, payload)
+				conn.ReadFrames(func(id int, streamType StreamType, payload []byte) {
+					require.Equal(t, 0, id)
+					require.Equal(t, StreamTypeRTP, streamType)
+					require.Equal(t, []byte{0x01, 0x02, 0x03, 0x04}, payload)
 
-				err = conn.WriteFrame(0, StreamTypeRTCP, []byte{0x05, 0x06, 0x07, 0x08})
+					err = conn.WriteFrame(0, StreamTypeRTCP, []byte{0x05, 0x06, 0x07, 0x08})
-				require.NoError(t, err)
+					require.NoError(t, err)
-			})
+				})
+			}()
 
 			<-frameRecv
 			conn.Close()
 			<-done
 
-			<-conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+			conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
 			})
 		})
 	}
@@ -590,9 +594,13 @@ func TestClientReadAnyPort(t *testing.T) {
 			require.NoError(t, err)
 
 			frameRecv := make(chan struct{})
-			done := conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+			done := make(chan struct{})
-				close(frameRecv)
+			go func() {
-			})
+				defer close(done)
+				conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+					close(frameRecv)
+				})
+			}()
 
 			<-frameRecv
 			conn.Close()
@@ -704,9 +712,13 @@ func TestClientReadAutomaticProtocol(t *testing.T) {
 		require.NoError(t, err)
 
 		frameRecv := make(chan struct{})
-		done := conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+		done := make(chan struct{})
-			close(frameRecv)
+		go func() {
-		})
+			defer close(done)
+			conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+				close(frameRecv)
+			})
+		}()
 
 		<-frameRecv
 		conn.Close()
@@ -872,9 +884,13 @@ func TestClientReadAutomaticProtocol(t *testing.T) {
 		require.NoError(t, err)
 
 		frameRecv := make(chan struct{})
-		done := conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+		done := make(chan struct{})
-			close(frameRecv)
+		go func() {
-		})
+			defer close(done)
+			conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+				close(frameRecv)
+			})
+		}()
 
 		<-frameRecv
 		conn.Close()
@@ -1013,9 +1029,13 @@ func TestClientReadRedirect(t *testing.T) {
 	require.NoError(t, err)
 
 	frameRecv := make(chan struct{})
-	done := conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+	done := make(chan struct{})
-		close(frameRecv)
+	go func() {
-	})
+		defer close(done)
+		conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+			close(frameRecv)
+		})
+	}()
 
 	<-frameRecv
 	conn.Close()
@@ -1214,18 +1234,22 @@ func TestClientReadPause(t *testing.T) {
 
 			firstFrame := int32(0)
 			frameRecv := make(chan struct{})
-			done := conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+			done := make(chan struct{})
-				if atomic.SwapInt32(&firstFrame, 1) == 0 {
+			go func() {
-					close(frameRecv)
+				defer close(done)
-				}
+				conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
-			})
+					if atomic.SwapInt32(&firstFrame, 1) == 0 {
+						close(frameRecv)
+					}
+				})
+			}()
 
 			<-frameRecv
 			_, err = conn.Pause()
 			require.NoError(t, err)
 			<-done
 
-			<-conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+			conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
 			})
 
 			_, err = conn.Play()
@@ -1233,11 +1257,15 @@ func TestClientReadPause(t *testing.T) {
 
 			firstFrame = int32(0)
 			frameRecv = make(chan struct{})
-			done = conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+			done = make(chan struct{})
-				if atomic.SwapInt32(&firstFrame, 1) == 0 {
+			go func() {
-					close(frameRecv)
+				defer close(done)
-				}
+				conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
-			})
+					if atomic.SwapInt32(&firstFrame, 1) == 0 {
+						close(frameRecv)
+					}
+				})
+			}()
 
 			<-frameRecv
 			conn.Close()
@@ -1398,12 +1426,16 @@ func TestClientReadRTCPReport(t *testing.T) {
 
 	recv := 0
 	recvDone := make(chan struct{})
-	done := conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
+	done := make(chan struct{})
-		recv++
+	go func() {
-		if recv >= 3 {
+		defer close(done)
-			close(recvDone)
+		conn.ReadFrames(func(id int, typ StreamType, payload []byte) {
-		}
+			recv++
-	})
+			if recv >= 3 {
+				close(recvDone)
+			}
+		})
+	}()
 
 	time.Sleep(1300 * time.Millisecond)
 
@@ -1559,7 +1591,7 @@ func TestClientReadErrorTimeout(t *testing.T) {
 			require.NoError(t, err)
 			defer conn.Close()
 
-			err = <-conn.ReadFrames(func(trackID int, streamType StreamType, payload []byte) {
+			err = conn.ReadFrames(func(trackID int, streamType StreamType, payload []byte) {
 			})
 
 			switch proto {
@@ -1688,13 +1720,17 @@ func TestClientReadIgnoreTCPInvalidTrack(t *testing.T) {
 
 	conn, err := c.DialRead("rtsp://localhost:8554/teststream")
 	require.NoError(t, err)
-	defer conn.Close()
 
 	recv := make(chan struct{})
-	conn.ReadFrames(func(trackID int, streamType StreamType, payload []byte) {
+	done := make(chan struct{})
-		close(recv)
+	go func() {
-	})
+		defer close(done)
-	require.NoError(t, err)
+		conn.ReadFrames(func(trackID int, streamType StreamType, payload []byte) {
+			close(recv)
+		})
+	}()
 
 	<-recv
+	conn.Close()
+	<-done
 }

clientconn.go

@@ -1164,7 +1164,7 @@ func (cc *ClientConn) doSetup(
 	}
 
 	proto := func() StreamProtocol {
-		// protocol set by previous Setup() or ReadFrames()
+		// protocol set by previous Setup() or switchProtocolIfTimeout()
 		if cc.streamProtocol != nil {
 			return *cc.streamProtocol
 		}
@@ -1566,8 +1566,7 @@ func (cc *ClientConn) Pause() (*base.Response, error) {
 }
 
 // ReadFrames starts reading frames.
-// it returns a channel that is written when the reading stops.
+func (cc *ClientConn) ReadFrames(onFrame func(int, StreamType, []byte)) error {
-func (cc *ClientConn) ReadFrames(onFrame func(int, StreamType, []byte)) chan error {
 	cc.readCBMutex.Lock()
 	cc.readCB = onFrame
 	cc.readCBMutex.Unlock()
@@ -1578,12 +1577,8 @@ func (cc *ClientConn) ReadFrames(onFrame func(int, StreamType, []byte)) chan err
 		cc.readCBSet = nil
 	}
 
-	ch := make(chan error, 1)
+	<-cc.backgroundDone
-	go func() {
+	return cc.backgroundErr
-		<-cc.backgroundDone
-		ch <- cc.backgroundErr
-	}()
-	return ch
 }
 
 // WriteFrame writes a frame.

examples/client-read-h264/main.go

@@ -38,7 +38,7 @@ func main() {
 	dec := rtph264.NewDecoder()
 
 	// read RTP frames
-	err = <-conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
+	err = conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
 		if trackID == h264Track {
 			// convert RTP frames into H264 NALUs
 			nalus, _, err := dec.Decode(buf)

examples/client-read-options/main.go

@@ -30,7 +30,7 @@ func main() {
 	defer conn.Close()
 
 	// read RTP frames
-	err = <-conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
+	err = conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
 		fmt.Printf("frame from track %d, type %v, size %d\n", trackID, typ, len(buf))
 	})
 	panic(err)

examples/client-read-partial/main.go

@@ -52,7 +52,7 @@ func main() {
 	}
 
 	// read RTP frames
-	err = <-conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
+	err = conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
 		fmt.Printf("frame from track %d, type %v, size %d\n", trackID, typ, len(buf))
 	})
 	panic(err)

examples/client-read-pause/main.go

@@ -23,9 +23,13 @@ func main() {
 
 	for {
 		// read RTP frames
-		done := conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
+		done := make(chan struct{})
-			fmt.Printf("frame from track %d, type %v, size %d\n", trackID, typ, len(buf))
+		go func() {
-		})
+			defer close(done)
+			conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
+				fmt.Printf("frame from track %d, type %v, size %d\n", trackID, typ, len(buf))
+			})
+		}()
 
 		// wait
 		time.Sleep(5 * time.Second)

examples/client-read/main.go

@@ -18,7 +18,7 @@ func main() {
 	defer conn.Close()
 
 	// read RTP frames
-	err = <-conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
+	err = conn.ReadFrames(func(trackID int, typ gortsplib.StreamType, buf []byte) {
 		fmt.Printf("frame from track %d, type %v, size %d\n", trackID, typ, len(buf))
 	})
 	panic(err)