diff --git a/adapter/experimental.go b/adapter/experimental.go index b4b9e6632..b5eb3243a 100644 --- a/adapter/experimental.go +++ b/adapter/experimental.go @@ -16,8 +16,7 @@ type ClashServer interface { Mode() string ModeList() []string SetMode(mode string) - SetModeUpdateHook(hook *observable.Subscriber[struct{}]) - HistoryStorage() URLTestHistoryStorage + AddModeUpdateHook(hook *observable.Subscriber[struct{}]) } type URLTestHistory struct { @@ -25,14 +24,6 @@ type URLTestHistory struct { Delay uint16 `json:"delay"` } -type URLTestHistoryStorage interface { - SetHook(hook *observable.Subscriber[struct{}]) - LoadURLTestHistory(tag string) *URLTestHistory - DeleteURLTestHistory(tag string) - StoreURLTestHistory(tag string, history *URLTestHistory) - Close() error -} - type V2RayServer interface { LifecycleService StatsService() ConnectionTracker diff --git a/box.go b/box.go index 1b8baf137..7e7b6b7a7 100644 --- a/box.go +++ b/box.go @@ -159,7 +159,7 @@ func New(options Options) (*Box, error) { needAPIService := common.Any(options.Services, func(it option.Service) bool { return it.Type == C.TypeAPI }) - if needAPIService && service.PtrFromContext[urltest.HistoryStorage](ctx) == nil { + if service.PtrFromContext[urltest.HistoryStorage](ctx) == nil { ctx = service.ContextWithPtr(ctx, urltest.NewHistoryStorage()) } platformInterface := service.FromContext[adapter.PlatformInterface](ctx) diff --git a/common/urltest/urltest.go b/common/urltest/urltest.go index 29d790e4d..11169c687 100644 --- a/common/urltest/urltest.go +++ b/common/urltest/urltest.go @@ -17,12 +17,10 @@ import ( "github.com/sagernet/sing/common/observable" ) -var _ adapter.URLTestHistoryStorage = (*HistoryStorage)(nil) - type HistoryStorage struct { access sync.RWMutex delayHistory map[string]*adapter.URLTestHistory - updateHook *observable.Subscriber[struct{}] + updateHooks []*observable.Subscriber[struct{}] } func NewHistoryStorage() *HistoryStorage { @@ -31,8 +29,16 @@ func NewHistoryStorage() *HistoryStorage { } } -func (s *HistoryStorage) SetHook(hook *observable.Subscriber[struct{}]) { - s.updateHook = hook +func (s *HistoryStorage) AddUpdateHook(hook *observable.Subscriber[struct{}]) { + s.access.Lock() + defer s.access.Unlock() + s.updateHooks = append(s.updateHooks, hook) +} + +func (s *HistoryStorage) NotifyUpdated() { + s.access.RLock() + defer s.access.RUnlock() + s.notifyUpdated() } func (s *HistoryStorage) LoadURLTestHistory(tag string) *adapter.URLTestHistory { @@ -59,8 +65,7 @@ func (s *HistoryStorage) StoreURLTestHistory(tag string, history *adapter.URLTes } func (s *HistoryStorage) notifyUpdated() { - updateHook := s.updateHook - if updateHook != nil { + for _, updateHook := range s.updateHooks { updateHook.Emit(struct{}{}) } } @@ -68,7 +73,7 @@ func (s *HistoryStorage) notifyUpdated() { func (s *HistoryStorage) Close() error { s.access.Lock() defer s.access.Unlock() - s.updateHook = nil + s.updateHooks = nil return nil } diff --git a/daemon/attached_service.go b/daemon/attached_service.go index bf965d904..b726ac2c5 100644 --- a/daemon/attached_service.go +++ b/daemon/attached_service.go @@ -19,9 +19,9 @@ func NewAttachedService(ctx context.Context) *StartedService { s.instance = instance s.serviceStatus = &ServiceStatus{Status: ServiceStatus_STARTED} s.startedAt = time.Now() - instance.urlTestHistoryStorage.SetHook(s.urlTestSubscriber) + instance.urlTestHistoryStorage.AddUpdateHook(s.urlTestSubscriber) if instance.clashServer != nil { - instance.clashServer.SetModeUpdateHook(s.clashModeSubscriber) + instance.clashServer.AddModeUpdateHook(s.clashModeSubscriber) } instance.logFactory.(log.ObservableFactory).AttachPlatformWriter(s) return s diff --git a/daemon/instance.go b/daemon/instance.go index 42c5b66b5..8ff58777e 100644 --- a/daemon/instance.go +++ b/daemon/instance.go @@ -28,7 +28,7 @@ type Instance struct { trafficManager *trafficcontrol.Manager cacheFile adapter.CacheFile pauseManager pause.Manager - urlTestHistoryStorage adapter.URLTestHistoryStorage + urlTestHistoryStorage *urltest.HistoryStorage outboundManager adapter.OutboundManager endpointManager adapter.EndpointManager logFactory log.Factory diff --git a/daemon/started_service.go b/daemon/started_service.go index 36298cd83..39da809eb 100644 --- a/daemon/started_service.go +++ b/daemon/started_service.go @@ -62,7 +62,6 @@ type StartedService struct { startedAt time.Time urlTestSubscriber *observable.Subscriber[struct{}] urlTestObserver *observable.Observer[struct{}] - urlTestHistoryStorage *urltest.HistoryStorage clashModeSubscriber *observable.Subscriber[struct{}] clashModeObserver *observable.Observer[struct{}] } @@ -102,7 +101,6 @@ func NewStartedService(options ServiceOptions) *StartedService { serviceStatusSubscriber: observable.NewSubscriber[*ServiceStatus](4), logSubscriber: observable.NewSubscriber[*log.Entry](128), urlTestSubscriber: observable.NewSubscriber[struct{}](1), - urlTestHistoryStorage: urltest.NewHistoryStorage(), clashModeSubscriber: observable.NewSubscriber[struct{}](1), } s.serviceStatusObserver = observable.NewObserver(s.serviceStatusSubscriber, 2) @@ -202,9 +200,9 @@ func (s *StartedService) StartOrReloadService(profileContent string, options *Ov return s.updateStatusError(err) } s.instance = instance - instance.urlTestHistoryStorage.SetHook(s.urlTestSubscriber) + instance.urlTestHistoryStorage.AddUpdateHook(s.urlTestSubscriber) if instance.clashServer != nil { - instance.clashServer.SetModeUpdateHook(s.clashModeSubscriber) + instance.clashServer.AddModeUpdateHook(s.clashModeSubscriber) } s.serviceAccess.Unlock() err = instance.Start() @@ -649,7 +647,6 @@ func (s *StartedService) SelectOutbound(ctx context.Context, request *SelectOutb if !selector.SelectOutbound(request.OutboundTag) { return nil, status.Error(codes.NotFound, "outbound not found in selector: "+request.OutboundTag) } - s.urlTestObserver.Emit(struct{}{}) return &emptypb.Empty{}, nil } diff --git a/experimental/clashapi/server.go b/experimental/clashapi/server.go index 149912d50..c04875366 100644 --- a/experimental/clashapi/server.go +++ b/experimental/clashapi/server.go @@ -9,6 +9,7 @@ import ( "os" "runtime" "strings" + "sync" "syscall" "time" @@ -49,12 +50,13 @@ type Server struct { logger log.Logger httpServer *http.Server trafficManager *trafficcontrol.Manager - urlTestHistory adapter.URLTestHistoryStorage + urlTestHistory *urltest.HistoryStorage logDebug bool - mode string - modeList []string - modeUpdateHook *observable.Subscriber[struct{}] + mode string + modeList []string + modeUpdateAccess sync.Mutex + modeUpdateHooks []*observable.Subscriber[struct{}] externalController bool externalUI string @@ -67,6 +69,10 @@ func NewServer(ctx context.Context, logFactory log.ObservableFactory, options op if trafficManager == nil { return nil, E.New("missing traffic manager") } + urlTestHistory := service.PtrFromContext[urltest.HistoryStorage](ctx) + if urlTestHistory == nil { + return nil, E.New("missing URL test history storage") + } chiRouter := chi.NewRouter() s := &Server{ ctx: ctx, @@ -81,16 +87,13 @@ func NewServer(ctx context.Context, logFactory log.ObservableFactory, options op Handler: chiRouter, }, trafficManager: trafficManager, + urlTestHistory: urlTestHistory, logDebug: logFactory.Level() >= log.LevelDebug, modeList: options.ModeList, externalController: options.ExternalController != "", externalUIDownloadURL: options.ExternalUIDownloadURL, externalUIDownloadDetour: options.ExternalUIDownloadDetour, } - s.urlTestHistory = service.FromContext[adapter.URLTestHistoryStorage](ctx) - if s.urlTestHistory == nil { - s.urlTestHistory = urltest.NewHistoryStorage() - } defaultMode := "Rule" if options.DefaultMode != "" { defaultMode = options.DefaultMode @@ -195,7 +198,6 @@ func (s *Server) Start(stage adapter.StartStage) error { func (s *Server) Close() error { return common.Close( common.PtrOrNil(s.httpServer), - s.urlTestHistory, ) } @@ -207,8 +209,10 @@ func (s *Server) ModeList() []string { return s.modeList } -func (s *Server) SetModeUpdateHook(hook *observable.Subscriber[struct{}]) { - s.modeUpdateHook = hook +func (s *Server) AddModeUpdateHook(hook *observable.Subscriber[struct{}]) { + s.modeUpdateAccess.Lock() + defer s.modeUpdateAccess.Unlock() + s.modeUpdateHooks = append(s.modeUpdateHooks, hook) } func (s *Server) SetMode(newMode string) { @@ -224,9 +228,11 @@ func (s *Server) SetMode(newMode string) { return } s.mode = newMode - if s.modeUpdateHook != nil { - s.modeUpdateHook.Emit(struct{}{}) + s.modeUpdateAccess.Lock() + for _, hook := range s.modeUpdateHooks { + hook.Emit(struct{}{}) } + s.modeUpdateAccess.Unlock() s.dnsRouter.ClearCache() cacheFile := service.FromContext[adapter.CacheFile](s.ctx) if cacheFile != nil { @@ -238,10 +244,6 @@ func (s *Server) SetMode(newMode string) { s.logger.Info("updated mode: ", newMode) } -func (s *Server) HistoryStorage() adapter.URLTestHistoryStorage { - return s.urlTestHistory -} - func authentication(serverSecret string) func(next http.Handler) http.Handler { return func(next http.Handler) http.Handler { fn := func(w http.ResponseWriter, r *http.Request) { diff --git a/protocol/group/selector.go b/protocol/group/selector.go index 85bea2b96..607387bd7 100644 --- a/protocol/group/selector.go +++ b/protocol/group/selector.go @@ -8,6 +8,7 @@ import ( "github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/adapter/outbound" "github.com/sagernet/sing-box/common/interrupt" + "github.com/sagernet/sing-box/common/urltest" C "github.com/sagernet/sing-box/constant" "github.com/sagernet/sing-box/log" "github.com/sagernet/sing-box/option" @@ -40,6 +41,7 @@ type Selector struct { defaultTag string outbounds map[string]adapter.Outbound selected common.TypedValue[adapter.Outbound] + history *urltest.HistoryStorage interruptGroup *interrupt.Group interruptExternalConnections bool } @@ -54,6 +56,7 @@ func NewSelector(ctx context.Context, router adapter.Router, logger log.ContextL tags: options.Outbounds, defaultTag: options.Default, outbounds: make(map[string]adapter.Outbound), + history: service.PtrFromContext[urltest.HistoryStorage](ctx), interruptGroup: interrupt.NewGroup(), interruptExternalConnections: options.InterruptExistConnections, } @@ -137,6 +140,9 @@ func (s *Selector) SelectOutbound(tag string) bool { } } s.interruptGroup.Interrupt(s.interruptExternalConnections) + if s.history != nil { + s.history.NotifyUpdated() + } return true } diff --git a/protocol/group/urltest.go b/protocol/group/urltest.go index 51fe3d0a4..bc13b6373 100644 --- a/protocol/group/urltest.go +++ b/protocol/group/urltest.go @@ -195,7 +195,7 @@ type URLTestGroup struct { interval time.Duration tolerance uint16 idleTimeout time.Duration - history adapter.URLTestHistoryStorage + history *urltest.HistoryStorage checking atomic.Bool selectedOutboundTCP adapter.Outbound selectedOutboundUDP adapter.Outbound @@ -221,13 +221,9 @@ func NewURLTestGroup(ctx context.Context, outboundManager adapter.OutboundManage if interval > idleTimeout { return nil, E.New("interval must be less or equal than idle_timeout") } - var history adapter.URLTestHistoryStorage - if historyFromCtx := service.PtrFromContext[urltest.HistoryStorage](ctx); historyFromCtx != nil { - history = historyFromCtx - } else if clashServer := service.FromContext[adapter.ClashServer](ctx); clashServer != nil { - history = clashServer.HistoryStorage() - } else { - history = urltest.NewHistoryStorage() + history := service.PtrFromContext[urltest.HistoryStorage](ctx) + if history == nil { + return nil, E.New("missing URL test history storage") } return &URLTestGroup{ ctx: ctx,