From c161b8e1c3f55a64c5a0c7271c295a844fa37b07 Mon Sep 17 00:00:00 2001
From: qydysky <32743305+qydysky@users.noreply.github.com>
Date: Thu, 13 Oct 2022 03:02:24 +0800
Subject: [PATCH] =?utf8?q?=E7=9B=B4=E6=92=AD=E6=B5=81=E6=81=A2=E5=A4=8D?=
 =?utf8?q?=E5=90=8E=E6=97=A0=E6=B3=95=E6=92=AD=E6=94=BE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=utf8
Content-Transfer-Encoding: 8bit

---
 Reply/F.go      |  5 +++--
 Reply/stream.go | 24 +++++++++++++++++++++++-
 2 files changed, 26 insertions(+), 3 deletions(-)

diff --git a/Reply/F.go b/Reply/F.go
index 6bbac68..04fac28 100644
--- a/Reply/F.go
+++ b/Reply/F.go
@@ -293,11 +293,12 @@ func init() {
 					tmp.LoadConfig(common, c.C.Log)
 					//关于ass的回调
 					tmp.Callback_start = func(ms *M4SStream) {
+						streamO.Store(ms.common.Roomid, tmp)                              //流服务添加
 						Ass_f(ms.Current_save_path, ms.Current_save_path+"0", time.Now()) //开始ass
 					}
 					tmp.Callback_stop = func(ms *M4SStream) {
-						streamO.Delete(ms.common.Roomid)
-						Ass_f("", "", time.Now()) //停止ass
+						streamO.Delete(ms.common.Roomid) //流服务去除
+						Ass_f("", "", time.Now())        //停止ass
 					}
 					if tmp.Start() {
 						streamO.Store(item.Roomid, tmp)
diff --git a/Reply/stream.go b/Reply/stream.go
index 57316ff..d47f0a4 100644
--- a/Reply/stream.go
+++ b/Reply/stream.go
@@ -486,12 +486,34 @@ func (t *M4SStream) saveStreamFlv() (e error) {
 			if err != nil {
 				out.Close()
 			}
+
 			rc, rw := io.Pipe()
+			var leastReadTime time.Time
+			// read timeout
+			go func() {
+				timer := time.NewTimer(5 * time.Second)
+				defer timer.Stop()
+				for {
+					select {
+					case <-s.WaitC():
+						return
+					case curT := <-timer.C:
+						if curT.Sub(leastReadTime).Seconds() > 5 {
+							// 5s未接收到任何数据
+							r.Cancel()
+							return
+						}
+					}
+				}
+			}()
+
+			// read
 			go func() {
 				var buff []byte
 				var buf = make([]byte, 1<<16)
 				for {
 					n, e := rc.Read(buf)
+					leastReadTime = time.Now()
 					buff = append(buff, buf[:n]...)
 					if n > 0 {
 						front_buf, keyframe, last_avilable_offset, e := Seach_stream_tag(buff)
@@ -869,7 +891,7 @@ func (t *M4SStream) Pusher(w http.ResponseWriter, r *http.Request) {
 	case `flv`:
 		t.pusherFlv(w, r)
 	default:
-		t.log.L(`E: `, `no support stream_type`)
+		t.log.L(`W: `, `Pusher no support stream_type`)
 	}
 }
 
-- 
2.39.2