👌 IMPROVE: 新增流关闭理由

http.go

@@ -122,18 +122,18 @@ func (conf *GlobalConfig) API_updateConfig(w http.ResponseWriter, r *http.Reques
 }
 
 func (conf *GlobalConfig) API_list_pull(w http.ResponseWriter, r *http.Request) {
-	var result []any
+	result := []any{}
 	Pullers.Range(func(key, value any) bool {
 		result = append(result, key)
 		return true
-	})	
+	})
 	if err := json.NewEncoder(w).Encode(result); err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 	}
 }
 
 func (conf *GlobalConfig) API_list_push(w http.ResponseWriter, r *http.Request) {
-	var result []any
+	result := []any{}
 	Pushers.Range(func(key, value any) bool {
 		result = append(result, key)
 		return true

io.go

@@ -71,7 +71,9 @@ func (i *IO[C]) OnEvent(event any) {
 		}
 	}
 }
-
+func (io *IO[C]) GetStream() *Stream {
+	return io.Stream
+}
 func (io *IO[C]) GetIO() *IO[C] {
 	return io
 }
@@ -86,6 +88,7 @@ type IIO interface {
 	Stop()
 	SetIO(any)
 	SetParentCtx(context.Context)
+	GetStream() *Stream
 }
 
 //Stop 停止订阅或者发布，由订阅者或者发布者调用

plugin.go

@@ -11,6 +11,7 @@ import (
 	"runtime"
 	"strings"
 	"sync"
+	"time"
 
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v3"
@@ -57,6 +58,7 @@ type Plugin struct {
 	RawConfig          config.Config //配置的map形式方便查询
 	Modified           config.Config //修改过的配置项
 	*zap.Logger        `json:"-"`
+	saveTimer          *time.Timer //用于保存的时候的延迟，防抖
 }
 
 func (opt *Plugin) logHandler(pattern string, handler func(http.ResponseWriter, *http.Request)) http.HandlerFunc {
@@ -172,15 +174,24 @@ func (opt *Plugin) settingPath() string {
 }
 
 func (opt *Plugin) Save() error {
-	file, err := os.OpenFile(opt.settingPath(), os.O_CREATE|os.O_WRONLY, 0644)
+	if opt.saveTimer == nil {
-	if err == nil {
+		var lock sync.Mutex
-		defer file.Close()
+		opt.saveTimer = time.AfterFunc(time.Second, func() {
-		err = yaml.NewEncoder(file).Encode(opt.Modified)
+			lock.Lock()
+			defer lock.Unlock()
+			file, err := os.OpenFile(opt.settingPath(), os.O_CREATE|os.O_WRONLY, 0644)
+			if err == nil {
+				defer file.Close()
+				err = yaml.NewEncoder(file).Encode(opt.Modified)
+			}
+			if err == nil {
+				opt.Info("config saved")
+			}
+		})
+	} else {
+		opt.saveTimer.Reset(time.Second)
 	}
-	if err == nil {
+	return nil
-		opt.Info("config saved")
-	}
-	return err
 }
 
 func (opt *Plugin) Publish(streamPath string, pub IPublisher) error {
@@ -233,7 +244,7 @@ func (opt *Plugin) Pull(streamPath string, url string, puller IPuller, save bool
 		defer opt.Info("stop pull", zap.String("remoteURL", url), zap.Error(err))
 		defer Pullers.Delete(puller)
 		for puller.Reconnect() {
-			if puller.Pull(); !puller.IsClosed() {
+			if puller.Pull(); puller.GetStream().IsShutdown() {
 				if err = puller.Connect(); err != nil {
 					return
 				}

stream.go

@@ -147,6 +147,7 @@ type Stream struct {
 	Tracks      Tracks
 	AppName     string
 	StreamName  string
+	CloseReason StreamAction //流关闭原因
 }
 type StreamSummay struct {
 	Path        string
@@ -240,6 +241,7 @@ func (r *Stream) action(action StreamAction) (ok bool) {
 			stateEvent = SEwaitClose{event}
 			r.timeout.Reset(r.DelayCloseTimeout)
 		case STATE_CLOSED:
+			r.CloseReason = action
 			for !r.actionChan.Close() {
 				// 等待channel发送完毕
 				time.Sleep(time.Millisecond * 100)
@@ -259,6 +261,11 @@ func (r *Stream) action(action StreamAction) (ok bool) {
 	}
 	return
 }
+
+func (r *Stream) IsShutdown() bool {
+	return r.CloseReason == ACTION_CLOSE
+}
+
 func (r *Stream) IsClosed() bool {
 	if r == nil {
 		return true