fix: Synchronize provider outbounds and update tickers

This commit is contained in:
Shtorm
2026-10-19 14:47:55 +03:00
parent 6b5ee55ba2
commit cb165a381c
3 changed files with 35 additions and 25 deletions
+20 -13
View File
@@ -28,6 +28,7 @@ type Adapter struct {
logger log.ContextLogger
providerType string
providerTag string
outboundAccess sync.RWMutex
outbounds []adapter.Outbound
outboundsByTag map[string]adapter.Outbound
ticker *time.Ticker
@@ -85,7 +86,10 @@ func (a *Adapter) Start() error {
if a.history == nil {
a.history = urltest.NewHistoryStorage()
}
go a.loopCheck()
if a.enabled {
a.ticker = time.NewTicker(a.interval)
go a.loopCheck()
}
return nil
}
@@ -98,13 +102,14 @@ func (a *Adapter) Tag() string {
}
func (a *Adapter) Outbounds() []adapter.Outbound {
a.outboundAccess.RLock()
defer a.outboundAccess.RUnlock()
return a.outbounds
}
func (a *Adapter) Outbound(tag string) (adapter.Outbound, bool) {
if a.outboundsByTag == nil {
return nil, false
}
a.outboundAccess.RLock()
defer a.outboundAccess.RUnlock()
detour, ok := a.outboundsByTag[tag]
return detour, ok
}
@@ -156,11 +161,13 @@ func (a *Adapter) UpdateOutbounds(oldOpts []option.Outbound, newOpts []option.Ou
outbounds = append(outbounds, outbound)
outboundsByTag[tag] = outbound
}
a.outboundAccess.Lock()
a.outbounds = outbounds
a.outboundsByTag = outboundsByTag
a.outboundAccess.Unlock()
if a.enabled && a.history != nil {
go a.HealthCheck(a.ctx)
}
a.outbounds = outbounds
a.outboundsByTag = outboundsByTag
}
func (a *Adapter) HealthCheck(ctx context.Context) (map[string]uint16, error) {
@@ -196,8 +203,11 @@ func (a *Adapter) Close() error {
if a.ticker != nil {
a.ticker.Stop()
}
a.outboundAccess.Lock()
outbounds := a.outbounds
a.outbounds = nil
a.outboundsByTag = nil
a.outboundAccess.Unlock()
var err error
for _, ob := range outbounds {
if err2 := a.outbound.Remove(ob.Tag()); err2 != nil {
@@ -210,10 +220,6 @@ func (a *Adapter) Close() error {
}
func (a *Adapter) loopCheck() {
if !a.enabled {
return
}
a.ticker = time.NewTicker(a.interval)
a.healthcheck(a.ctx)
for {
select {
@@ -234,7 +240,7 @@ func (a *Adapter) healthcheck(ctx context.Context) (map[string]uint16, error) {
b, _ := batch.New(ctx, batch.WithConcurrencyNum[any](10))
var resultAccess sync.Mutex
checked := make(map[string]bool)
for _, detour := range a.outbounds {
for _, detour := range a.Outbounds() {
tag := detour.Tag()
if checked[tag] {
continue
@@ -265,7 +271,8 @@ func (a *Adapter) healthcheck(ctx context.Context) (map[string]uint16, error) {
}
func (a *Adapter) removeUseless(newOpts []option.Outbound) {
if len(a.outbounds) == 0 {
outbounds := a.Outbounds()
if len(outbounds) == 0 {
return
}
exists := make(map[string]bool)
@@ -278,7 +285,7 @@ func (a *Adapter) removeUseless(newOpts []option.Outbound) {
}
exists[tag] = true
}
for _, opt := range a.outbounds {
for _, opt := range outbounds {
if !exists[opt.Tag()] {
if err := a.outbound.Remove(opt.Tag()); err != nil {
a.logger.Error(err, "close outbound [", opt.Tag(), "]")
+7 -3
View File
@@ -102,13 +102,17 @@ func (s *ProviderLocal) Start(stage adapter.StartStage, scope *adapter.Scope) er
return err
}
s.UpdateGroups()
err = s.Adapter.Start()
if err != nil {
return err
}
if s.watcher != nil {
err := s.watcher.Start()
err = s.watcher.Start()
if err != nil {
s.logger.Error(E.Cause(err, "watch provider file"))
}
}
return s.Adapter.Start()
return nil
}
func (s *ProviderLocal) UpdatedAt() time.Time {
@@ -133,5 +137,5 @@ func (s *ProviderLocal) reloadFile(path string) error {
}
func (s *ProviderLocal) Close() error {
return common.Close(&s.Adapter, common.PtrOrNil(s.watcher))
return common.Close(common.PtrOrNil(s.watcher), &s.Adapter)
}
+8 -9
View File
@@ -135,9 +135,13 @@ func (s *ProviderRemote) Start(stage adapter.StartStage, scope *adapter.Scope) e
} else {
s.dialer = s.outbound.Default()
}
err := s.Adapter.Start()
if err != nil {
return err
}
s.ticker = time.NewTicker(s.updateInterval)
go s.loopUpdate()
return s.Adapter.Start()
return nil
}
func (s *ProviderRemote) References() []string {
@@ -289,22 +293,17 @@ func (s *ProviderRemote) fetch(ctx context.Context) error {
func (s *ProviderRemote) loopUpdate() {
if time.Since(s.lastUpdated) < s.updateInterval {
select {
case <-s.ctx.Done():
return
case <-time.After(time.Until(s.lastUpdated.Add(s.updateInterval))):
s.updateOnce()
}
s.ticker.Reset(time.Until(s.lastUpdated.Add(s.updateInterval)))
} else {
s.updateOnce()
}
s.ticker = time.NewTicker(s.updateInterval)
for {
runtime.GC()
select {
case <-s.ctx.Done():
return
case <-s.ticker.C:
s.ticker.Reset(s.updateInterval)
s.updateOnce()
}
}