优化代码

internal/nodes/api_stream.go

@@ -41,7 +41,7 @@ func NewAPIStream() *APIStream {
 }
 
 func (this *APIStream) Start() {
-	events.On(events.EventQuit, func() {
+	events.OnKey(events.EventQuit, this, func() {
 		this.isQuiting = true
 		if this.cancelFunc != nil {
 			this.cancelFunc()

internal/nodes/client_conn_traffic.go

@@ -16,7 +16,7 @@ import (
 // 发送监控流量
 func init() {
 	events.On(events.EventStart, func() {
-		ticker := time.NewTicker(1 * time.Minute)
+		var ticker = time.NewTicker(1 * time.Minute)
 		goman.New(func() {
 			for range ticker.C {
 				// 加入到数据队列中

internal/nodes/listener.go

@@ -61,7 +61,7 @@ func (this *Listener) listenTCP() error {
 		return err
 	}
 	var netListener = NewClientListener(tcpListener, protocol.IsHTTPFamily() || protocol.IsHTTPSFamily())
-	events.On(events.EventQuit, func() {
+	events.OnKey(events.EventQuit, this, func() {
 		remotelogs.Println("LISTENER", "quit "+this.group.FullAddr())
 		_ = netListener.Close()
 	})
@@ -122,7 +122,7 @@ func (this *Listener) listenUDP() error {
 	if err != nil {
 		return err
 	}
-	events.On(events.EventQuit, func() {
+	events.OnKey(events.EventQuit, this, func() {
 		remotelogs.Println("LISTENER", "quit "+this.group.FullAddr())
 		_ = listener.Close()
 	})
@@ -143,6 +143,8 @@ func (this *Listener) listenUDP() error {
 }
 
 func (this *Listener) Close() error {
+	events.Remove(this)
+
 	if this.listener == nil {
 		return nil
 	}

internal/nodes/listener_base.go

@@ -25,8 +25,8 @@ func (this *BaseListener) Reset() {
 
 }
 
-// CountActiveListeners 获取当前活跃连接数
-func (this *BaseListener) CountActiveListeners() int {
+// CountActiveConnections 获取当前活跃连接数
+func (this *BaseListener) CountActiveConnections() int {
 	return types.Int(this.countActiveConnections)
 }
 

internal/nodes/listener_interface.go

@@ -16,6 +16,6 @@ type ListenerInterface interface {
 	// Reload 重载配置
 	Reload(serverGroup *serverconfigs.ServerAddressGroup)
 
-	// CountActiveListeners 获取当前活跃的连接数
-	CountActiveListeners() int
+	// CountActiveConnections 获取当前活跃的连接数
+	CountActiveConnections() int
 }

internal/nodes/listener_manager.go

@@ -159,7 +159,7 @@ func (this *ListenerManager) TotalActiveConnections() int {
 
 	total := 0
 	for _, listener := range this.listenersMap {
-		total += listener.listener.CountActiveListeners()
+		total += listener.listener.CountActiveConnections()
 	}
 	return total
 }

internal/nodes/node.go

@@ -451,7 +451,7 @@ func (this *Node) syncConfig(taskVersion int64) error {
 func (this *Node) startSyncTimer() {
 	// TODO 这个时间间隔可以自行设置
 	ticker := time.NewTicker(60 * time.Second)
-	events.On(events.EventQuit, func() {
+	events.OnKey(events.EventQuit, this, func() {
 		remotelogs.Println("NODE", "quit sync timer")
 		ticker.Stop()
 	})
@@ -691,7 +691,7 @@ func (this *Node) listenSock() error {
 		}
 	})
 
-	events.On(events.EventQuit, func() {
+	events.OnKey(events.EventQuit, this, func() {
 		logs.Println("NODE", "quit unix sock")
 		_ = this.sock.Close()
 	})

internal/nodes/node_status_executor.go

@@ -41,9 +41,9 @@ func (this *NodeStatusExecutor) Listen() {
 	this.update()
 
 	// TODO 这个时间间隔可以配置
-	ticker := time.NewTicker(30 * time.Second)
+	var ticker = time.NewTicker(30 * time.Second)
 
-	events.On(events.EventQuit, func() {
+	events.OnKey(events.EventQuit, this, func() {
 		remotelogs.Println("NODE_STATUS", "quit executor")
 		ticker.Stop()
 	})

internal/nodes/origin_state_manager.go

@@ -21,6 +21,9 @@ func init() {
 			SharedOriginStateManager.Start()
 		})
 	})
+	events.On(events.EventQuit, func() {
+		SharedOriginStateManager.Stop()
+	})
 }
 
 // OriginStateManager 源站状态管理
@@ -41,7 +44,7 @@ func NewOriginStateManager() *OriginStateManager {
 
 // Start 启动
 func (this *OriginStateManager) Start() {
-	events.On(events.EventReload, func() {
+	events.OnKey(events.EventReload, this, func() {
 		this.locker.Lock()
 		this.stateMap = map[int64]*OriginState{}
 		this.locker.Unlock()
@@ -58,6 +61,12 @@ func (this *OriginStateManager) Start() {
 	}
 }
 
+func (this *OriginStateManager) Stop() {
+	if this.ticker != nil {
+		this.ticker.Stop()
+	}
+}
+
 // Loop 单次循环检查
 func (this *OriginStateManager) Loop() error {
 	if sharedNodeConfig == nil {

internal/nodes/system_services.go

@@ -28,7 +28,7 @@ func init() {
 	})
 }
 
-// 系统服务管理
+// SystemServiceManager 系统服务管理
 type SystemServiceManager struct {
 }
 

internal/nodes/task_sync_api_nodes.go

@@ -7,7 +7,6 @@ import (
 	"github.com/TeaOSLab/EdgeNode/internal/configs"
 	"github.com/TeaOSLab/EdgeNode/internal/events"
 	"github.com/TeaOSLab/EdgeNode/internal/goman"
-	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeNode/internal/rpc"
 	"github.com/TeaOSLab/EdgeNode/internal/trackers"
 	"github.com/iwind/TeaGo/Tea"
@@ -21,17 +20,22 @@ import (
 	"time"
 )
 
+var sharedSyncAPINodesTask = NewSyncAPINodesTask()
+
 func init() {
 	events.On(events.EventStart, func() {
-		task := NewSyncAPINodesTask()
 		goman.New(func() {
-			task.Start()
+			sharedSyncAPINodesTask.Start()
 		})
 	})
+	events.On(events.EventQuit, func() {
+		sharedSyncAPINodesTask.Stop()
+	})
 }
 
 // SyncAPINodesTask API节点同步任务
 type SyncAPINodesTask struct {
+	ticker *time.Ticker
 }
 
 func NewSyncAPINodesTask() *SyncAPINodesTask {
@@ -39,16 +43,12 @@ func NewSyncAPINodesTask() *SyncAPINodesTask {
 }
 
 func (this *SyncAPINodesTask) Start() {
-	ticker := time.NewTicker(5 * time.Minute)
+	this.ticker = time.NewTicker(5 * time.Minute)
 	if Tea.IsTesting() {
 		// 快速测试
-		ticker = time.NewTicker(1 * time.Minute)
+		this.ticker = time.NewTicker(1 * time.Minute)
 	}
-	events.On(events.EventQuit, func() {
-		remotelogs.Println("SYNC_API_NODES_TASK", "quit task")
-		ticker.Stop()
-	})
-	for range ticker.C {
+	for range this.ticker.C {
 		err := this.Loop()
 		if err != nil {
 			logs.Println("[TASK][SYNC_API_NODES_TASK]" + err.Error())
@@ -56,6 +56,12 @@ func (this *SyncAPINodesTask) Start() {
 	}
 }
 
+func (this *SyncAPINodesTask) Stop() {
+	if this.ticker != nil {
+		this.ticker.Stop()
+	}
+}
+
 func (this *SyncAPINodesTask) Loop() error {
 	var tr = trackers.Begin("SYNC_API_NODES")
 	defer tr.End()