Merge tag 'v1.14.0'

experimental/clashapi/api_meta.go

@@ -5,10 +5,10 @@ import (
 	"context"
 	"net"
 	"net/http"
+	"runtime"
 	"runtime/debug"
 	"time"
 
-	"github.com/sagernet/sing-box/experimental/clashapi/trafficontrol"
 	"github.com/sagernet/sing/common/json"
 	"github.com/sagernet/ws"
 	"github.com/sagernet/ws/wsutil"
@@ -28,7 +28,7 @@ func (s *Server) setupMetaAPI(r chi.Router) {
 		})
 		r.Mount("/", middleware.Profiler())
 	}
-	r.Get("/memory", memory(s.ctx, s.trafficManager))
+	r.Get("/memory", memory(s.ctx))
 	r.Mount("/group", groupRouter(s))
 	r.Mount("/upgrade", upgradeRouter(s))
 }
@@ -38,7 +38,13 @@ type Memory struct {
 	OSLimit uint64 `json:"oslimit"` // maybe we need it in the future
 }
 
-func memory(ctx context.Context, trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
+func inuseMemory() uint64 {
+	var memStats runtime.MemStats
+	runtime.ReadMemStats(&memStats)
+	return memStats.StackInuse + memStats.HeapInuse + memStats.HeapIdle - memStats.HeapReleased
+}
+
+func memory(ctx context.Context) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		var conn net.Conn
 		if r.Header.Get("Upgrade") == "websocket" {
@@ -68,7 +74,7 @@ func memory(ctx context.Context, trafficManager *trafficontrol.Manager) func(w h
 			}
 			buf.Reset()
 
-			inuse := trafficManager.Snapshot().Memory
+			inuse := inuseMemory()
 
 			// make chat.js begin with zero
 			// this is shit var,but we need output 0 for first time

experimental/clashapi/api_meta_group.go

@@ -5,14 +5,11 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/sagernet/sing-box/adapter"
-	"github.com/sagernet/sing-box/common/urltest"
 	"github.com/sagernet/sing-box/protocol/group"
 	"github.com/sagernet/sing/common"
-	"github.com/sagernet/sing/common/batch"
 	"github.com/sagernet/sing/common/json/badjson"
 
 	"github.com/go-chi/chi/v5"
@@ -89,40 +86,7 @@ func getGroupDelay(server *Server) func(w http.ResponseWriter, r *http.Request)
 				itOutbound, _ := server.outbound.Outbound(it)
 				return itOutbound
 			}))
-			b, _ := batch.New(ctx, batch.WithConcurrencyNum[any](10))
-			checked := make(map[string]bool)
-			result = make(map[string]uint16)
-			var resultAccess sync.Mutex
-			for _, detour := range outbounds {
-				tag := detour.Tag()
-				realTag := group.RealTag(detour)
-				if checked[realTag] {
-					continue
-				}
-				checked[realTag] = true
-				p, loaded := server.outbound.Outbound(realTag)
-				if !loaded {
-					continue
-				}
-				b.Go(realTag, func() (any, error) {
-					t, err := urltest.URLTest(ctx, url, p)
-					if err != nil {
-						server.logger.Debug("outbound ", tag, " unavailable: ", err)
-						server.urlTestHistory.DeleteURLTestHistory(realTag)
-					} else {
-						server.logger.Debug("outbound ", tag, " available: ", t, "ms")
-						server.urlTestHistory.StoreURLTestHistory(realTag, &adapter.URLTestHistory{
-							Time:  time.Now(),
-							Delay: t,
-						})
-						resultAccess.Lock()
-						result[tag] = t
-						resultAccess.Unlock()
-					}
-					return nil, nil
-				})
-			}
-			b.Wait()
+			result = group.URLTestOutbounds(ctx, server.outbound, server.urlTestHistory, server.logger, outbounds, url, 0, true)
 		}
 
 		if err != nil {

experimental/clashapi/connections.go

@@ -8,7 +8,10 @@ import (
 	"time"
 
 	"github.com/sagernet/sing-box/adapter"
-	"github.com/sagernet/sing-box/experimental/clashapi/trafficontrol"
+	"github.com/sagernet/sing-box/common/trafficcontrol"
+	C "github.com/sagernet/sing-box/constant"
+	"github.com/sagernet/sing/common"
+	F "github.com/sagernet/sing/common/format"
 	"github.com/sagernet/sing/common/json"
 	"github.com/sagernet/ws"
 	"github.com/sagernet/ws/wsutil"
@@ -18,19 +21,93 @@ import (
 	"github.com/gofrs/uuid/v5"
 )
 
-func connectionRouter(ctx context.Context, router adapter.Router, trafficManager *trafficontrol.Manager) http.Handler {
+func connectionRouter(ctx context.Context, network adapter.NetworkManager, trafficManager *trafficcontrol.Manager) http.Handler {
 	r := chi.NewRouter()
 	r.Get("/", getConnections(ctx, trafficManager))
-	r.Delete("/", closeAllConnections(router, trafficManager))
+	r.Delete("/", closeAllConnections(ctx, network, trafficManager))
 	r.Delete("/{id}", closeConnection(trafficManager))
 	return r
 }
 
-func getConnections(ctx context.Context, trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
+func connectionsSnapshot(trafficManager *trafficcontrol.Manager) render.M {
+	uplinkTotal, downlinkTotal := trafficManager.Total()
+	connections := common.Filter(trafficManager.Connections(), func(metadata *trafficcontrol.TrackerMetadata) bool {
+		return metadata.OutboundType != C.TypeDNS
+	})
+	return render.M{
+		"downloadTotal": downlinkTotal,
+		"uploadTotal":   uplinkTotal,
+		"connections": common.Map(connections, func(metadata *trafficcontrol.TrackerMetadata) connectionObject {
+			return connectionObject(*metadata)
+		}),
+		"memory": inuseMemory(),
+	}
+}
+
+type connectionObject trafficcontrol.TrackerMetadata
+
+func (c connectionObject) MarshalJSON() ([]byte, error) {
+	var inbound string
+	if c.Metadata.Inbound != "" {
+		inbound = c.Metadata.InboundType + "/" + c.Metadata.Inbound
+	} else {
+		inbound = c.Metadata.InboundType
+	}
+	var domain string
+	if c.Metadata.Domain != "" {
+		domain = c.Metadata.Domain
+	} else {
+		domain = c.Metadata.Destination.Fqdn
+	}
+	var processPath string
+	if c.Metadata.ProcessInfo != nil {
+		if c.Metadata.ProcessInfo.ProcessPath != "" {
+			processPath = c.Metadata.ProcessInfo.ProcessPath
+		} else if len(c.Metadata.ProcessInfo.AndroidPackageNames) > 0 {
+			processPath = c.Metadata.ProcessInfo.AndroidPackageNames[0]
+		}
+		if processPath == "" {
+			if c.Metadata.ProcessInfo.UserId != -1 {
+				processPath = F.ToString(c.Metadata.ProcessInfo.UserId)
+			}
+		} else if c.Metadata.ProcessInfo.UserName != "" {
+			processPath = F.ToString(processPath, " (", c.Metadata.ProcessInfo.UserName, ")")
+		} else if c.Metadata.ProcessInfo.UserId != -1 {
+			processPath = F.ToString(processPath, " (", c.Metadata.ProcessInfo.UserId, ")")
+		}
+	}
+	var rule string
+	if c.Rule != nil {
+		rule = F.ToString(c.Rule, " => ", c.Rule.Action())
+	} else {
+		rule = "final"
+	}
+	return json.Marshal(map[string]any{
+		"id": c.ID,
+		"metadata": map[string]any{
+			"network":         c.Metadata.Network,
+			"type":            inbound,
+			"sourceIP":        c.Metadata.Source.Addr,
+			"destinationIP":   c.Metadata.Destination.Addr,
+			"sourcePort":      F.ToString(c.Metadata.Source.Port),
+			"destinationPort": F.ToString(c.Metadata.Destination.Port),
+			"host":            domain,
+			"dnsMode":         "normal",
+			"processPath":     processPath,
+		},
+		"upload":      c.Upload.Load(),
+		"download":    c.Download.Load(),
+		"start":       c.CreatedAt,
+		"chains":      c.Chain,
+		"rule":        rule,
+		"rulePayload": "",
+	})
+}
+
+func getConnections(ctx context.Context, trafficManager *trafficcontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		if r.Header.Get("Upgrade") != "websocket" {
-			snapshot := trafficManager.Snapshot()
-			render.JSON(w, r, snapshot)
+			render.JSON(w, r, connectionsSnapshot(trafficManager))
 			return
 		}
 
@@ -56,9 +133,9 @@ func getConnections(ctx context.Context, trafficManager *trafficontrol.Manager)
 		buf := &bytes.Buffer{}
 		sendSnapshot := func() error {
 			buf.Reset()
-			snapshot := trafficManager.Snapshot()
-			if err := json.NewEncoder(buf).Encode(snapshot); err != nil {
-				return err
+			encodeErr := json.NewEncoder(buf).Encode(connectionsSnapshot(trafficManager))
+			if encodeErr != nil {
+				return encodeErr
 			}
 			return wsutil.WriteServerText(conn, buf.Bytes())
 		}
@@ -82,27 +159,21 @@ func getConnections(ctx context.Context, trafficManager *trafficontrol.Manager)
 	}
 }
 
-func closeConnection(trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
+func closeConnection(trafficManager *trafficcontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		id := uuid.FromStringOrNil(chi.URLParam(r, "id"))
-		snapshot := trafficManager.Snapshot()
-		for _, c := range snapshot.Connections {
-			if id == c.Metadata().ID {
-				c.Close()
-				break
-			}
+		targetConnection := trafficManager.Connection(id)
+		if targetConnection != nil {
+			targetConnection.Close()
 		}
 		render.NoContent(w, r)
 	}
 }
 
-func closeAllConnections(router adapter.Router, trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
+func closeAllConnections(ctx context.Context, network adapter.NetworkManager, trafficManager *trafficcontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
-		snapshot := trafficManager.Snapshot()
-		for _, c := range snapshot.Connections {
-			c.Close()
-		}
-		router.ResetNetwork()
+		trafficManager.CloseAllConnections()
+		network.ResetNetwork(ctx)
 		render.NoContent(w, r)
 	}
 }

experimental/clashapi/proxies.go

@@ -70,7 +70,7 @@ func proxyInfo(server *Server, detour adapter.Outbound) *badjson.JSONObject {
 	info.Put("type", clashType)
 	info.Put("name", detour.Tag())
 	info.Put("udp", common.Contains(detour.Network(), N.NetworkUDP))
-	delayHistory := server.urlTestHistory.LoadURLTestHistory(adapter.OutboundTag(detour))
+	delayHistory := server.urlTestHistory.LoadURLTestHistory(group.RealTag(server.outbound, detour))
 	if delayHistory != nil {
 		info.Put("history", []*adapter.URLTestHistory{delayHistory})
 	} else {
@@ -184,6 +184,30 @@ func updateProxy(w http.ResponseWriter, r *http.Request) {
 	render.NoContent(w, r)
 }
 
+func groupContains(outboundManager adapter.OutboundManager, outboundGroup adapter.OutboundGroup, tag string, visited map[string]bool) bool {
+	for _, memberTag := range outboundGroup.All() {
+		if memberTag == tag {
+			return true
+		}
+		member, loaded := outboundManager.Outbound(memberTag)
+		if !loaded {
+			continue
+		}
+		if group.RealTag(outboundManager, member) == tag {
+			return true
+		}
+		memberGroup, isGroup := member.(adapter.OutboundGroup)
+		if !isGroup || visited[memberTag] {
+			continue
+		}
+		visited[memberTag] = true
+		if groupContains(outboundManager, memberGroup, tag, visited) {
+			return true
+		}
+	}
+	return false
+}
+
 func getProxyDelay(server *Server) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		query := r.URL.Query()
@@ -204,7 +228,7 @@ func getProxyDelay(server *Server) func(w http.ResponseWriter, r *http.Request)
 
 		delay, err := urltest.URLTest(ctx, url, proxy)
 		defer func() {
-			realTag := group.RealTag(proxy)
+			realTag := group.RealTag(server.outbound, proxy)
 			if err != nil {
 				server.urlTestHistory.DeleteURLTestHistory(realTag)
 			} else {
@@ -213,6 +237,16 @@ func getProxyDelay(server *Server) func(w http.ResponseWriter, r *http.Request)
 					Delay: delay,
 				})
 			}
+			for _, detour := range server.outbound.Outbounds() {
+				urlTestGroup, isURLTestGroup := detour.(adapter.URLTestGroup)
+				if !isURLTestGroup {
+					continue
+				}
+				if !groupContains(server.outbound, urlTestGroup, realTag, map[string]bool{detour.Tag(): true}) {
+					continue
+				}
+				urlTestGroup.PerformUpdateCheck()
+			}
 		}()
 
 		if ctx.Err() != nil {

experimental/clashapi/server.go

@@ -9,21 +9,21 @@ import (
 	"os"
 	"runtime"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
 	"github.com/sagernet/cors"
 	"github.com/sagernet/sing-box/adapter"
+	"github.com/sagernet/sing-box/common/trafficcontrol"
 	"github.com/sagernet/sing-box/common/urltest"
 	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing-box/experimental"
-	"github.com/sagernet/sing-box/experimental/clashapi/trafficontrol"
 	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-box/option"
 	"github.com/sagernet/sing/common"
 	E "github.com/sagernet/sing/common/exceptions"
 	"github.com/sagernet/sing/common/json"
-	N "github.com/sagernet/sing/common/network"
 	"github.com/sagernet/sing/common/observable"
 	"github.com/sagernet/sing/service"
 	"github.com/sagernet/sing/service/filemanager"
@@ -42,6 +42,7 @@ var _ adapter.ClashServer = (*Server)(nil)
 
 type Server struct {
 	ctx            context.Context
+	network        adapter.NetworkManager
 	router         adapter.Router
 	dnsRouter      adapter.DNSRouter
 	outbound       adapter.OutboundManager
@@ -49,13 +50,14 @@ type Server struct {
 	provider       adapter.ProviderManager
 	logger         log.Logger
 	httpServer     *http.Server
-	trafficManager *trafficontrol.Manager
-	urlTestHistory adapter.URLTestHistoryStorage
+	trafficManager *trafficcontrol.Manager
+	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
@@ -64,10 +66,18 @@ type Server struct {
 }
 
 func NewServer(ctx context.Context, logFactory log.ObservableFactory, options option.ClashAPIOptions) (adapter.ClashServer, error) {
-	trafficManager := trafficontrol.NewManager()
+	trafficManager := service.PtrFromContext[trafficcontrol.Manager](ctx)
+	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,
+		network:   service.FromContext[adapter.NetworkManager](ctx),
 		router:    service.FromContext[adapter.Router](ctx),
 		dnsRouter: service.FromContext[adapter.DNSRouter](ctx),
 		outbound:  service.FromContext[adapter.OutboundManager](ctx),
@@ -79,16 +89,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
@@ -123,7 +130,7 @@ func NewServer(ctx context.Context, logFactory log.ObservableFactory, options op
 		r.Mount("/configs", configRouter(s, logFactory))
 		r.Mount("/proxies", proxyRouter(s, s.router))
 		r.Mount("/rules", ruleRouter(s.router))
-		r.Mount("/connections", connectionRouter(s.ctx, s.router, trafficManager))
+		r.Mount("/connections", connectionRouter(s.ctx, s.network, trafficManager))
 		r.Mount("/providers/proxies", proxyProviderRouter(s))
 		r.Mount("/providers/rules", ruleProviderRouter())
 		r.Mount("/script", scriptRouter())
@@ -135,6 +142,10 @@ func NewServer(ctx context.Context, logFactory log.ObservableFactory, options op
 	})
 	if options.ExternalUI != "" {
 		s.externalUI = filemanager.BasePath(ctx, os.ExpandEnv(options.ExternalUI))
+		_, err := filemanager.ReadDir(ctx, s.externalUI)
+		if err != nil && !os.IsNotExist(err) {
+			return nil, E.Cause(err, "read external UI directory")
+		}
 		chiRouter.Group(func(r chi.Router) {
 			r.Get("/ui", http.RedirectHandler("/ui/", http.StatusMovedPermanently).ServeHTTP)
 			r.Handle("/ui/*", http.StripPrefix("/ui/", http.FileServer(Dir(s.externalUI))))
@@ -193,8 +204,6 @@ func (s *Server) Start(stage adapter.StartStage) error {
 func (s *Server) Close() error {
 	return common.Close(
 		common.PtrOrNil(s.httpServer),
-		s.trafficManager,
-		s.urlTestHistory,
 	)
 }
 
@@ -206,8 +215,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) {
@@ -223,9 +234,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 {
@@ -237,22 +250,6 @@ func (s *Server) SetMode(newMode string) {
 	s.logger.Notice("updated mode: ", newMode)
 }
 
-func (s *Server) HistoryStorage() adapter.URLTestHistoryStorage {
-	return s.urlTestHistory
-}
-
-func (s *Server) TrafficManager() *trafficontrol.Manager {
-	return s.trafficManager
-}
-
-func (s *Server) RoutedConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, matchedRule adapter.Rule, matchOutbound adapter.Outbound) net.Conn {
-	return trafficontrol.NewTCPTracker(conn, s.trafficManager, metadata, s.outbound, matchedRule, matchOutbound)
-}
-
-func (s *Server) RoutedPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, matchedRule adapter.Rule, matchOutbound adapter.Outbound) N.PacketConn {
-	return trafficontrol.NewUDPTracker(conn, s.trafficManager, metadata, s.outbound, matchedRule, matchOutbound)
-}
-
 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) {
@@ -305,7 +302,7 @@ type Traffic struct {
 	Down int64 `json:"down"`
 }
 
-func traffic(ctx context.Context, trafficManager *trafficontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
+func traffic(ctx context.Context, trafficManager *trafficcontrol.Manager) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		var conn net.Conn
 		if r.Header.Get("Upgrade") == "websocket" {

experimental/clashapi/server_resources.go

@@ -7,7 +7,6 @@ import (
 	"io"
 	"net"
 	"net/http"
-	"os"
 	"path/filepath"
 	"strings"
 
@@ -24,9 +23,9 @@ func (s *Server) checkAndDownloadExternalUI() {
 	if s.externalUI == "" {
 		return
 	}
-	entries, err := os.ReadDir(s.externalUI)
+	entries, err := filemanager.ReadDir(s.ctx, s.externalUI)
 	if err != nil {
-		os.MkdirAll(s.externalUI, 0o755)
+		filemanager.MkdirAll(s.ctx, s.externalUI, 0o755)
 	}
 	if len(entries) == 0 {
 		err = s.downloadExternalUI()
@@ -79,7 +78,7 @@ func (s *Server) downloadExternalUI() error {
 	}
 	err = s.downloadZIP(response.Body, s.externalUI)
 	if err != nil {
-		removeAllInDirectory(s.externalUI)
+		removeAllInDirectory(s.ctx, s.externalUI)
 	}
 	return err
 }
@@ -89,7 +88,7 @@ func (s *Server) downloadZIP(body io.Reader, output string) error {
 	if err != nil {
 		return err
 	}
-	defer os.Remove(tempFile.Name())
+	defer filemanager.Remove(s.ctx, tempFile.Name())
 	_, err = io.Copy(tempFile, body)
 	tempFile.Close()
 	if err != nil {
@@ -113,7 +112,7 @@ func (s *Server) downloadZIP(body io.Reader, output string) error {
 		if len(pathElements) > 1 {
 			saveDirectory = filepath.Join(saveDirectory, filepath.Join(pathElements[:len(pathElements)-1]...))
 		}
-		err = os.MkdirAll(saveDirectory, 0o755)
+		err = filemanager.MkdirAll(s.ctx, saveDirectory, 0o755)
 		if err != nil {
 			return err
 		}
@@ -140,13 +139,13 @@ func downloadZIPEntry(ctx context.Context, zipFile *zip.File, savePath string) e
 	return common.Error(io.Copy(saveFile, reader))
 }
 
-func removeAllInDirectory(directory string) {
-	dirEntries, err := os.ReadDir(directory)
+func removeAllInDirectory(ctx context.Context, directory string) {
+	dirEntries, err := filemanager.ReadDir(ctx, directory)
 	if err != nil {
 		return
 	}
 	for _, dirEntry := range dirEntries {
-		os.RemoveAll(filepath.Join(directory, dirEntry.Name()))
+		filemanager.RemoveAll(ctx, filepath.Join(directory, dirEntry.Name()))
 	}
 }
 

experimental/clashapi/trafficontrol/manager.go

@@ -1,181 +0,0 @@
-package trafficontrol
-
-import (
-	"runtime"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/sagernet/sing-box/common/compatible"
-	C "github.com/sagernet/sing-box/constant"
-	"github.com/sagernet/sing/common"
-	"github.com/sagernet/sing/common/json"
-	"github.com/sagernet/sing/common/observable"
-	"github.com/sagernet/sing/common/x/list"
-
-	"github.com/gofrs/uuid/v5"
-)
-
-type ConnectionEventType int
-
-const (
-	ConnectionEventNew ConnectionEventType = iota
-	ConnectionEventUpdate
-	ConnectionEventClosed
-)
-
-type ConnectionEvent struct {
-	Type          ConnectionEventType
-	ID            uuid.UUID
-	Metadata      *TrackerMetadata
-	UplinkDelta   int64
-	DownlinkDelta int64
-	ClosedAt      time.Time
-}
-
-const closedConnectionsLimit = 1000
-
-type Manager struct {
-	uploadTotal   atomic.Int64
-	downloadTotal atomic.Int64
-
-	connections             compatible.Map[uuid.UUID, Tracker]
-	closedConnectionsAccess sync.Mutex
-	closedConnections       list.List[TrackerMetadata]
-	memory                  uint64
-
-	eventSubscriber *observable.Subscriber[ConnectionEvent]
-}
-
-func NewManager() *Manager {
-	return &Manager{}
-}
-
-func (m *Manager) SetEventHook(subscriber *observable.Subscriber[ConnectionEvent]) {
-	m.eventSubscriber = subscriber
-}
-
-func (m *Manager) Join(c Tracker) {
-	metadata := c.Metadata()
-	m.connections.Store(metadata.ID, c)
-	if m.eventSubscriber != nil {
-		m.eventSubscriber.Emit(ConnectionEvent{
-			Type:     ConnectionEventNew,
-			ID:       metadata.ID,
-			Metadata: metadata,
-		})
-	}
-}
-
-func (m *Manager) Leave(c Tracker) {
-	metadata := c.Metadata()
-	_, loaded := m.connections.LoadAndDelete(metadata.ID)
-	if loaded {
-		closedAt := time.Now()
-		metadata.ClosedAt = closedAt
-		metadataCopy := *metadata
-		m.closedConnectionsAccess.Lock()
-		if m.closedConnections.Len() >= closedConnectionsLimit {
-			m.closedConnections.PopFront()
-		}
-		m.closedConnections.PushBack(metadataCopy)
-		m.closedConnectionsAccess.Unlock()
-		if m.eventSubscriber != nil {
-			m.eventSubscriber.Emit(ConnectionEvent{
-				Type:     ConnectionEventClosed,
-				ID:       metadata.ID,
-				Metadata: &metadataCopy,
-				ClosedAt: closedAt,
-			})
-		}
-	}
-}
-
-func (m *Manager) PushUploaded(size int64) {
-	m.uploadTotal.Add(size)
-}
-
-func (m *Manager) PushDownloaded(size int64) {
-	m.downloadTotal.Add(size)
-}
-
-func (m *Manager) Total() (up int64, down int64) {
-	return m.uploadTotal.Load(), m.downloadTotal.Load()
-}
-
-func (m *Manager) ConnectionsLen() int {
-	return m.connections.Len()
-}
-
-func (m *Manager) Connections() []*TrackerMetadata {
-	var connections []*TrackerMetadata
-	m.connections.Range(func(_ uuid.UUID, value Tracker) bool {
-		connections = append(connections, value.Metadata())
-		return true
-	})
-	return connections
-}
-
-func (m *Manager) ClosedConnections() []*TrackerMetadata {
-	m.closedConnectionsAccess.Lock()
-	values := m.closedConnections.Array()
-	m.closedConnectionsAccess.Unlock()
-	if len(values) == 0 {
-		return nil
-	}
-	connections := make([]*TrackerMetadata, len(values))
-	for i := range values {
-		connections[i] = &values[i]
-	}
-	return connections
-}
-
-func (m *Manager) Connection(id uuid.UUID) Tracker {
-	connection, loaded := m.connections.Load(id)
-	if !loaded {
-		return nil
-	}
-	return connection
-}
-
-func (m *Manager) Snapshot() *Snapshot {
-	var connections []Tracker
-	m.connections.Range(func(_ uuid.UUID, value Tracker) bool {
-		if value.Metadata().OutboundType != C.TypeDNS {
-			connections = append(connections, value)
-		}
-		return true
-	})
-
-	var memStats runtime.MemStats
-	runtime.ReadMemStats(&memStats)
-	m.memory = memStats.StackInuse + memStats.HeapInuse + memStats.HeapIdle - memStats.HeapReleased
-
-	return &Snapshot{
-		Upload:      m.uploadTotal.Load(),
-		Download:    m.downloadTotal.Load(),
-		Connections: connections,
-		Memory:      m.memory,
-	}
-}
-
-func (m *Manager) ResetStatistic() {
-	m.uploadTotal.Store(0)
-	m.downloadTotal.Store(0)
-}
-
-type Snapshot struct {
-	Download    int64
-	Upload      int64
-	Connections []Tracker
-	Memory      uint64
-}
-
-func (s *Snapshot) MarshalJSON() ([]byte, error) {
-	return json.Marshal(map[string]any{
-		"downloadTotal": s.Download,
-		"uploadTotal":   s.Upload,
-		"connections":   common.Map(s.Connections, func(t Tracker) *TrackerMetadata { return t.Metadata() }),
-		"memory":        s.Memory,
-	})
-}

experimental/clashapi/trafficontrol/tracker.go

@@ -1,254 +0,0 @@
-package trafficontrol
-
-import (
-	"net"
-	"sync/atomic"
-	"time"
-
-	"github.com/sagernet/sing-box/adapter"
-	"github.com/sagernet/sing/common"
-	"github.com/sagernet/sing/common/bufio"
-	F "github.com/sagernet/sing/common/format"
-	"github.com/sagernet/sing/common/json"
-	N "github.com/sagernet/sing/common/network"
-
-	"github.com/gofrs/uuid/v5"
-)
-
-type TrackerMetadata struct {
-	ID           uuid.UUID
-	Metadata     adapter.InboundContext
-	CreatedAt    time.Time
-	ClosedAt     time.Time
-	Upload       *atomic.Int64
-	Download     *atomic.Int64
-	Chain        []string
-	Rule         adapter.Rule
-	Outbound     string
-	OutboundType string
-}
-
-func (t TrackerMetadata) MarshalJSON() ([]byte, error) {
-	var inbound string
-	if t.Metadata.Inbound != "" {
-		inbound = t.Metadata.InboundType + "/" + t.Metadata.Inbound
-	} else {
-		inbound = t.Metadata.InboundType
-	}
-	var domain string
-	if t.Metadata.Domain != "" {
-		domain = t.Metadata.Domain
-	} else {
-		domain = t.Metadata.Destination.Fqdn
-	}
-	var processPath string
-	if t.Metadata.ProcessInfo != nil {
-		if t.Metadata.ProcessInfo.ProcessPath != "" {
-			processPath = t.Metadata.ProcessInfo.ProcessPath
-		} else if len(t.Metadata.ProcessInfo.AndroidPackageNames) > 0 {
-			processPath = t.Metadata.ProcessInfo.AndroidPackageNames[0]
-		}
-		if processPath == "" {
-			if t.Metadata.ProcessInfo.UserId != -1 {
-				processPath = F.ToString(t.Metadata.ProcessInfo.UserId)
-			}
-		} else if t.Metadata.ProcessInfo.UserName != "" {
-			processPath = F.ToString(processPath, " (", t.Metadata.ProcessInfo.UserName, ")")
-		} else if t.Metadata.ProcessInfo.UserId != -1 {
-			processPath = F.ToString(processPath, " (", t.Metadata.ProcessInfo.UserId, ")")
-		}
-	}
-	var rule string
-	if t.Rule != nil {
-		rule = F.ToString(t.Rule, " => ", t.Rule.Action())
-	} else {
-		rule = "final"
-	}
-	return json.Marshal(map[string]any{
-		"id": t.ID,
-		"metadata": map[string]any{
-			"network":         t.Metadata.Network,
-			"type":            inbound,
-			"sourceIP":        t.Metadata.Source.Addr,
-			"destinationIP":   t.Metadata.Destination.Addr,
-			"sourcePort":      F.ToString(t.Metadata.Source.Port),
-			"destinationPort": F.ToString(t.Metadata.Destination.Port),
-			"host":            domain,
-			"dnsMode":         "normal",
-			"processPath":     processPath,
-		},
-		"upload":      t.Upload.Load(),
-		"download":    t.Download.Load(),
-		"start":       t.CreatedAt,
-		"chains":      t.Chain,
-		"rule":        rule,
-		"rulePayload": "",
-	})
-}
-
-type Tracker interface {
-	Metadata() *TrackerMetadata
-	Close() error
-}
-
-type TCPConn struct {
-	N.ExtendedConn
-	metadata TrackerMetadata
-	manager  *Manager
-}
-
-func (tt *TCPConn) Metadata() *TrackerMetadata {
-	return &tt.metadata
-}
-
-func (tt *TCPConn) Close() error {
-	tt.manager.Leave(tt)
-	return tt.ExtendedConn.Close()
-}
-
-func (tt *TCPConn) Upstream() any {
-	return tt.ExtendedConn
-}
-
-func (tt *TCPConn) ReaderReplaceable() bool {
-	return true
-}
-
-func (tt *TCPConn) WriterReplaceable() bool {
-	return true
-}
-
-func NewTCPTracker(conn net.Conn, manager *Manager, metadata adapter.InboundContext, outboundManager adapter.OutboundManager, matchRule adapter.Rule, matchOutbound adapter.Outbound) *TCPConn {
-	id, _ := uuid.NewV4()
-	var (
-		chain        []string
-		next         string
-		outbound     string
-		outboundType string
-	)
-	if matchOutbound != nil {
-		next = matchOutbound.Tag()
-	} else {
-		next = outboundManager.Default().Tag()
-	}
-	for {
-		detour, loaded := outboundManager.Outbound(next)
-		if !loaded {
-			break
-		}
-		chain = append(chain, next)
-		outbound = detour.Tag()
-		outboundType = detour.Type()
-		group, isGroup := detour.(adapter.OutboundGroup)
-		if !isGroup {
-			break
-		}
-		next = group.Now()
-	}
-	upload := new(atomic.Int64)
-	download := new(atomic.Int64)
-	tracker := &TCPConn{
-		ExtendedConn: bufio.NewCounterConn(conn, []N.CountFunc{func(n int64) {
-			upload.Add(n)
-			manager.PushUploaded(n)
-		}}, []N.CountFunc{func(n int64) {
-			download.Add(n)
-			manager.PushDownloaded(n)
-		}}),
-		metadata: TrackerMetadata{
-			ID:           id,
-			Metadata:     metadata,
-			CreatedAt:    time.Now(),
-			Upload:       upload,
-			Download:     download,
-			Chain:        common.Reverse(chain),
-			Rule:         matchRule,
-			Outbound:     outbound,
-			OutboundType: outboundType,
-		},
-		manager: manager,
-	}
-	manager.Join(tracker)
-	return tracker
-}
-
-type UDPConn struct {
-	N.PacketConn `json:"-"`
-	metadata     TrackerMetadata
-	manager      *Manager
-}
-
-func (ut *UDPConn) Metadata() *TrackerMetadata {
-	return &ut.metadata
-}
-
-func (ut *UDPConn) Close() error {
-	ut.manager.Leave(ut)
-	return ut.PacketConn.Close()
-}
-
-func (ut *UDPConn) Upstream() any {
-	return ut.PacketConn
-}
-
-func (ut *UDPConn) ReaderReplaceable() bool {
-	return true
-}
-
-func (ut *UDPConn) WriterReplaceable() bool {
-	return true
-}
-
-func NewUDPTracker(conn N.PacketConn, manager *Manager, metadata adapter.InboundContext, outboundManager adapter.OutboundManager, matchRule adapter.Rule, matchOutbound adapter.Outbound) *UDPConn {
-	id, _ := uuid.NewV4()
-	var (
-		chain        []string
-		next         string
-		outbound     string
-		outboundType string
-	)
-	if matchOutbound != nil {
-		next = matchOutbound.Tag()
-	} else {
-		next = outboundManager.Default().Tag()
-	}
-	for {
-		detour, loaded := outboundManager.Outbound(next)
-		if !loaded {
-			break
-		}
-		chain = append(chain, next)
-		outbound = detour.Tag()
-		outboundType = detour.Type()
-		group, isGroup := detour.(adapter.OutboundGroup)
-		if !isGroup {
-			break
-		}
-		next = group.Now()
-	}
-	upload := new(atomic.Int64)
-	download := new(atomic.Int64)
-	trackerConn := &UDPConn{
-		PacketConn: bufio.NewCounterPacketConn(conn, []N.CountFunc{func(n int64) {
-			upload.Add(n)
-			manager.PushUploaded(n)
-		}}, []N.CountFunc{func(n int64) {
-			download.Add(n)
-			manager.PushDownloaded(n)
-		}}),
-		metadata: TrackerMetadata{
-			ID:           id,
-			Metadata:     metadata,
-			CreatedAt:    time.Now(),
-			Upload:       upload,
-			Download:     download,
-			Chain:        common.Reverse(chain),
-			Rule:         matchRule,
-			Outbound:     outbound,
-			OutboundType: outboundType,
-		},
-		manager: manager,
-	}
-	manager.Join(trackerConn)
-	return trackerConn
-}