apps/engine
1
0
Fork 0
mirror of https://github.com/Monibuca/engine.git synced 2025-10-05 16:46:58 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: delayclosetimeout not work

Browse Source
This commit is contained in:
langhuihui
2024-08-29 19:51:03 +08:00
parent 005fb482f8
commit f231a0cd71
1 changed files with 18 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
stream.go
View File

@@ -295,6 +295,13 @@ func findOrCreateStream(streamPath string, waitTimeout time.Duration) (s *Stream
} }
} }
func (r *Stream) resetTimer(dur time.Duration) {
if r.timeout != nil {
r.Warn("reset timer", zap.Duration("timeout", dur))
r.timeout.Reset(dur)
}
}
func (r *Stream) action(action StreamAction) (ok bool) { func (r *Stream) action(action StreamAction) (ok bool) {
var event StateEvent var event StateEvent
event.Target = r event.Target = r
@@ -327,7 +334,7 @@ func (r *Stream) action(action StreamAction) (ok bool) {
} else if waitTime == 0 { } else if waitTime == 0 {
waitTime = time.Millisecond * 10 //没有订阅者也没有配置发布者等待重连时间，默认10ms后关闭流 waitTime = time.Millisecond * 10 //没有订阅者也没有配置发布者等待重连时间，默认10ms后关闭流
} }
r.timeout.Reset(waitTime) r.resetTimer(waitTime)
r.Debug("wait publisher", zap.Duration("wait timeout", waitTime)) r.Debug("wait publisher", zap.Duration("wait timeout", waitTime))
case STATE_WAITTRACK: case STATE_WAITTRACK:
if len(r.SEHistory) > 1 { if len(r.SEHistory) > 1 {
@@ -335,7 +342,7 @@ func (r *Stream) action(action StreamAction) (ok bool) {
} else { } else {
stateEvent = SEpublish{event} stateEvent = SEpublish{event}
} }
r.timeout.Reset(time.Second * 5) // 5秒心跳，检测track的存活度 r.resetTimer(time.Second * 5) // 5秒心跳，检测track的存活度
case STATE_PUBLISHING: case STATE_PUBLISHING:
stateEvent = SEtrackAvaliable{event} stateEvent = SEtrackAvaliable{event}
r.Subscribers.SendInviteTrack(r) r.Subscribers.SendInviteTrack(r)
@@ -343,13 +350,13 @@ func (r *Stream) action(action StreamAction) (ok bool) {
if puller, ok := r.Publisher.(IPuller); ok { if puller, ok := r.Publisher.(IPuller); ok {
puller.OnConnected() puller.OnConnected()
} }
r.timeout.Reset(time.Second * 5) // 5秒心跳，检测track的存活度 r.resetTimer(time.Second * 5) // 5秒心跳，检测track的存活度
case STATE_WAITCLOSE: case STATE_WAITCLOSE:
stateEvent = SEwaitClose{event} stateEvent = SEwaitClose{event}
if r.IdleTimeout > 0 { if r.IdleTimeout > 0 {
r.timeout.Reset(r.IdleTimeout) r.resetTimer(r.IdleTimeout)
} else { } else {
r.timeout.Reset(r.DelayCloseTimeout) r.resetTimer(r.DelayCloseTimeout)
} }
case STATE_CLOSED: case STATE_CLOSED:
Streams.Delete(r.Path) Streams.Delete(r.Path)
@@ -508,11 +515,14 @@ func (s *Stream) run() {
continue continue
} }
} }
if s.State == STATE_WAITTRACK { switch s.State {
case STATE_WAITTRACK:
s.action(ACTION_TRACKAVAILABLE) s.action(ACTION_TRACKAVAILABLE)
case STATE_WAITCLOSE:
continue
} }
s.Subscribers.AbortWait() s.Subscribers.AbortWait()
s.timeout.Reset(time.Second * 5) s.resetTimer(time.Second * 5)
} else { } else {
s.Debug("timeout", timeOutInfo) s.Debug("timeout", timeOutInfo)
s.action(ACTION_TIMEOUT) s.action(ACTION_TIMEOUT)