Compare commits

...
Author SHA1 Message Date
Fangliding 8ec957cc72 refactor 2026-10-27 16:03:46 +08:00
+32 -31
View File
@@ -3,11 +3,7 @@ package signal
import ( import (
"context" "context"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/xtls/xray-core/common"
"github.com/xtls/xray-core/common/task"
) )
type ActivityUpdater interface { type ActivityUpdater interface {
@@ -17,10 +13,10 @@ type ActivityUpdater interface {
type ActivityTimer struct { type ActivityTimer struct {
mu sync.RWMutex mu sync.RWMutex
updated chan struct{} updated chan struct{}
checkTask *task.Periodic checker *time.Timer
timeout time.Duration
onTimeout func() onTimeout func()
consumed atomic.Bool consumed bool
once sync.Once
} }
func (t *ActivityTimer) Update() { func (t *ActivityTimer) Update() {
@@ -30,30 +26,29 @@ func (t *ActivityTimer) Update() {
} }
} }
func (t *ActivityTimer) check() error { func (t *ActivityTimer) check() bool {
select { select {
case <-t.updated: case <-t.updated:
return true
default: default:
t.finish() t.finish()
return false
} }
return nil
} }
func (t *ActivityTimer) finish() { func (t *ActivityTimer) finish() {
t.once.Do(func() { t.mu.Lock()
t.consumed.Store(true) if t.consumed {
t.mu.Lock() t.mu.Unlock()
defer t.mu.Unlock() return
}
common.CloseIfExists(t.checkTask) t.checker.Stop()
t.onTimeout() t.consumed = true
}) t.mu.Unlock()
t.onTimeout()
} }
func (t *ActivityTimer) SetTimeout(timeout time.Duration) { func (t *ActivityTimer) SetTimeout(timeout time.Duration) {
if t.consumed.Load() {
return
}
if timeout == 0 { if timeout == 0 {
t.finish() t.finish()
return return
@@ -61,25 +56,31 @@ func (t *ActivityTimer) SetTimeout(timeout time.Duration) {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
// double check, just in case if t.consumed {
if t.consumed.Load() {
return return
} }
newCheckTask := &task.Periodic{ t.timeout = timeout
Interval: timeout, t.checker.Reset(timeout)
Execute: t.check,
}
common.CloseIfExists(t.checkTask)
t.checkTask = newCheckTask
t.Update()
common.Must(newCheckTask.Start())
} }
func CancelAfterInactivity(ctx context.Context, cancel context.CancelFunc, timeout time.Duration) *ActivityTimer { func CancelAfterInactivity(ctx context.Context, cancel context.CancelFunc, timeout time.Duration) *ActivityTimer {
timer := &ActivityTimer{ timer := &ActivityTimer{
updated: make(chan struct{}, 1), updated: make(chan struct{}, 1),
onTimeout: cancel, onTimeout: cancel,
timeout: timeout,
}
timer.checker = time.AfterFunc(timeout, func() {
if timer.check() {
timer.mu.Lock()
defer timer.mu.Unlock()
if timer.consumed {
return
}
timer.checker.Reset(timer.timeout)
}
})
if timeout == 0 {
timer.SetTimeout(0)
} }
timer.SetTimeout(timeout)
return timer return timer
} }