Merge pull request '修复调试问题' (#7) from feature_gxh into master

agent/internal/services/mq/cache.go

@@ -147,9 +147,9 @@ func (svc *Service) WaitCacheMovePackage(msg *agtmq.WaitCacheMovePackage) (*agtm
 				errMsg = tsk.Error().Error()
 			}
 
-			return mq.ReplyOK(agtmq.NewWaitCacheMovePackageResp(true, errMsg, nil))
+			return mq.ReplyOK(agtmq.NewWaitCacheMovePackageResp(true, errMsg, mvPkgTask.ResultCacheInfos))
 		}
 
-		return mq.ReplyOK(agtmq.NewWaitCacheMovePackageResp(false, "", mvPkgTask.ResultCacheInfos))
+		return mq.ReplyOK(agtmq.NewWaitCacheMovePackageResp(false, "", nil))
 	}
 }

agent/internal/services/mq/storage.go

@@ -27,7 +27,7 @@ func (svc *Service) StartStorageLoadPackage(msg *agtmq.StartStorageLoadPackage)
 
 		return nil, mq.Failed(errorcode.OperationFailed, "new coordinator client failed")
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	getStgResp, err := coorCli.GetStorageInfo(coormq.NewGetStorageInfo(msg.UserID, msg.StorageID))
 	if err != nil {
@@ -163,7 +163,7 @@ func (svc *Service) StartStorageCreatePackage(msg *agtmq.StartStorageCreatePacka
 
 		return nil, mq.Failed(errorcode.OperationFailed, "new coordinator client failed")
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	getStgResp, err := coorCli.GetStorageInfo(coormq.NewGetStorageInfo(msg.UserID, msg.StorageID))
 	if err != nil {

agent/internal/task/cache_move_package.go

@@ -61,7 +61,7 @@ func (t *CacheMovePackage) do(ctx TaskContext) error {
 	if err != nil {
 		return fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	pkgResp, err := coorCli.GetPackage(coormq.NewGetPackage(t.userID, t.packageID))
 	if err != nil {
@@ -71,12 +71,13 @@ func (t *CacheMovePackage) do(ctx TaskContext) error {
 	if pkgResp.Redundancy.IsRepInfo() {
 		return t.moveRep(ctx, coorCli, pkgResp.Package)
 	} else {
+		return fmt.Errorf("not implement yet!")
 		// TODO EC的CacheMove逻辑
 	}
 
 	return nil
 }
-func (t *CacheMovePackage) moveRep(ctx TaskContext, coorCli *coormq.PoolClient, pkg model.Package) error {
+func (t *CacheMovePackage) moveRep(ctx TaskContext, coorCli *coormq.Client, pkg model.Package) error {
 	getRepResp, err := coorCli.GetPackageObjectRepData(coormq.NewGetPackageObjectRepData(pkg.PackageID))
 	if err != nil {
 		return fmt.Errorf("getting package object rep data: %w", err)

agent/main.go

@@ -69,9 +69,10 @@ func main() {
 	if err != nil {
 		log.Fatalf("new agent server failed, err: %s", err.Error())
 	}
-	agtSvr.OnError = func(err error) {
+	agtSvr.OnError(func(err error) {
 		log.Warnf("agent server err: %s", err.Error())
-	}
+	})
+
 	go serveAgentServer(agtSvr, &wg)
 
 	go reportStatus(&wg) //网络延迟感知

client/internal/services/bucket.go

@@ -27,7 +27,7 @@ func (svc *BucketService) GetUserBuckets(userID int64) ([]model.Bucket, error) {
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	resp, err := coorCli.GetUserBuckets(coormq.NewGetUserBuckets(userID))
 	if err != nil {
@@ -42,7 +42,7 @@ func (svc *BucketService) GetBucketPackages(userID int64, bucketID int64) ([]mod
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	resp, err := coorCli.GetBucketPackages(coormq.NewGetBucketPackages(userID, bucketID))
 	if err != nil {
@@ -57,7 +57,7 @@ func (svc *BucketService) CreateBucket(userID int64, bucketName string) (int64,
 	if err != nil {
 		return 0, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	// TODO 只有阅读了系统操作的源码，才能知道要加哪些锁，但用户的命令可能会调用不止一个系统操作。
 	// 因此加锁的操作还是必须在用户命令里完成，但具体加锁的内容，则需要被封装起来与系统操作放到一起，方便管理，避免分散改动。
@@ -85,7 +85,7 @@ func (svc *BucketService) DeleteBucket(userID int64, bucketID int64) error {
 	if err != nil {
 		return fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	// TODO 检查用户是否有删除这个Bucket的权限。检查的时候可以只上UserBucket的Read锁
 

client/internal/services/cacah.go

@@ -23,7 +23,7 @@ func (svc *CacheService) StartCacheMovePackage(userID int64, packageID int64, no
 	if err != nil {
 		return "", fmt.Errorf("new agent client: %w", err)
 	}
-	defer agentCli.Close()
+	defer stgglb.AgentMQPool.Release(agentCli)
 
 	startResp, err := agentCli.StartCacheMovePackage(agtmq.NewStartCacheMovePackage(userID, packageID))
 	if err != nil {
@@ -38,7 +38,7 @@ func (svc *CacheService) WaitCacheMovePackage(nodeID int64, taskID string, waitT
 	if err != nil {
 		return true, nil, fmt.Errorf("new agent client: %w", err)
 	}
-	defer agentCli.Close()
+	defer stgglb.AgentMQPool.Release(agentCli)
 
 	waitResp, err := agentCli.WaitCacheMovePackage(agtmq.NewWaitCacheMovePackage(taskID, waitTimeout.Milliseconds()))
 	if err != nil {

client/internal/services/package.go

@@ -28,7 +28,7 @@ func (svc *PackageService) DownloadPackage(userID int64, packageID int64) (itera
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	mutex, err := reqbuilder.NewBuilder().
 		// 用于判断用户是否有对象权限
@@ -89,7 +89,7 @@ func (svc *PackageService) DownloadPackage(userID int64, packageID int64) (itera
 	}
 }
 
-func (svc *PackageService) downloadRepPackage(packageID int64, objects []model.Object, coorCli *coormq.PoolClient) (*iterator.RepObjectIterator, error) {
+func (svc *PackageService) downloadRepPackage(packageID int64, objects []model.Object, coorCli *coormq.Client) (*iterator.RepObjectIterator, error) {
 	getObjRepDataResp, err := coorCli.GetPackageObjectRepData(coormq.NewGetPackageObjectRepData(packageID))
 	if err != nil {
 		return nil, fmt.Errorf("getting package object rep data: %w", err)
@@ -101,7 +101,7 @@ func (svc *PackageService) downloadRepPackage(packageID int64, objects []model.O
 
 	return iter, nil
 }
-func (svc *PackageService) downloadECPackage(pkg model.Package, objects []model.Object, coorCli *coormq.PoolClient) (*iterator.ECObjectIterator, error) {
+func (svc *PackageService) downloadECPackage(pkg model.Package, objects []model.Object, coorCli *coormq.Client) (*iterator.ECObjectIterator, error) {
 	getObjECDataResp, err := coorCli.GetPackageObjectECData(coormq.NewGetPackageObjectECData(pkg.PackageID))
 	if err != nil {
 		return nil, fmt.Errorf("getting package object ec data: %w", err)
@@ -185,7 +185,7 @@ func (svc *PackageService) DeletePackage(userID int64, packageID int64) error {
 	if err != nil {
 		return fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	mutex, err := reqbuilder.NewBuilder().
 		Metadata().
@@ -220,7 +220,7 @@ func (svc *PackageService) GetCachedNodes(userID int64, packageID int64) (stgsdk
 	if err != nil {
 		return stgsdk.PackageCachingInfo{}, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	resp, err := coorCli.GetPackageCachedNodes(coormq.NewGetPackageCachedNodes(userID, packageID))
 	if err != nil {
@@ -240,7 +240,7 @@ func (svc *PackageService) GetLoadedNodes(userID int64, packageID int64) ([]int6
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	resp, err := coorCli.GetPackageLoadedNodes(coormq.NewGetPackageLoadedNodes(userID, packageID))
 	if err != nil {

client/internal/services/scanner.go

@@ -21,7 +21,7 @@ func (svc *ScannerService) PostEvent(event scevt.Event, isEmergency bool, dontMe
 	if err != nil {
 		return fmt.Errorf("new scacnner client: %w", err)
 	}
-	defer scCli.Close()
+	defer stgglb.ScannerMQPool.Release(scCli)
 
 	err = scCli.PostEvent(scmq.NewPostEvent(event, isEmergency, dontMerge))
 	if err != nil {

client/internal/services/storage.go

@@ -44,7 +44,7 @@ func (svc *StorageService) StartStorageCreatePackage(userID int64, bucketID int6
 	if err != nil {
 		return 0, "", fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	stgResp, err := coorCli.GetStorageInfo(coormq.NewGetStorageInfo(userID, storageID))
 	if err != nil {
@@ -55,7 +55,7 @@ func (svc *StorageService) StartStorageCreatePackage(userID int64, bucketID int6
 	if err != nil {
 		return 0, "", fmt.Errorf("new agent client: %w", err)
 	}
-	defer agentCli.Close()
+	defer stgglb.AgentMQPool.Release(agentCli)
 
 	startResp, err := agentCli.StartStorageCreatePackage(agtmq.NewStartStorageCreatePackage(userID, bucketID, name, storageID, path, redundancy, nodeAffinity))
 	if err != nil {
@@ -71,7 +71,7 @@ func (svc *StorageService) WaitStorageCreatePackage(nodeID int64, taskID string,
 		// TODO 失败是否要当做任务已经结束？
 		return true, 0, fmt.Errorf("new agent client: %w", err)
 	}
-	defer agentCli.Close()
+	defer stgglb.AgentMQPool.Release(agentCli)
 
 	waitResp, err := agentCli.WaitStorageCreatePackage(agtmq.NewWaitStorageCreatePackage(taskID, waitTimeout.Milliseconds()))
 	if err != nil {

client/internal/task/storage_load_package.go

@@ -62,7 +62,7 @@ func (t *StorageLoadPackage) do(ctx TaskContext) error {
 	if err != nil {
 		return fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	getStgResp, err := coorCli.GetStorageInfo(coormq.NewGetStorageInfo(t.userID, t.storageID))
 	if err != nil {
@@ -70,13 +70,13 @@ func (t *StorageLoadPackage) do(ctx TaskContext) error {
 	}
 
 	// 然后向代理端发送移动文件的请求
-	agentClient, err := stgglb.AgentMQPool.Acquire(getStgResp.NodeID)
+	agentCli, err := stgglb.AgentMQPool.Acquire(getStgResp.NodeID)
 	if err != nil {
 		return fmt.Errorf("create agent client to %d failed, err: %w", getStgResp.NodeID, err)
 	}
-	defer agentClient.Close()
+	defer stgglb.AgentMQPool.Release(agentCli)
 
-	agentMoveResp, err := agentClient.StartStorageLoadPackage(
+	agentMoveResp, err := agentCli.StartStorageLoadPackage(
 		agtmq.NewStartStorageLoadPackage(
 			t.userID,
 			t.packageID,
@@ -87,7 +87,7 @@ func (t *StorageLoadPackage) do(ctx TaskContext) error {
 	}
 
 	for {
-		waitResp, err := agentClient.WaitStorageLoadPackage(agtmq.NewWaitStorageLoadPackage(agentMoveResp.TaskID, int64(time.Second)*5))
+		waitResp, err := agentCli.WaitStorageLoadPackage(agtmq.NewWaitStorageLoadPackage(agentMoveResp.TaskID, int64(time.Second)*5))
 		if err != nil {
 			return fmt.Errorf("wait loading package: %w", err)
 		}

common/globals/pools.go

@@ -9,11 +9,11 @@ import (
 	scmq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/scanner"
 )
 
-var AgentMQPool *agtmq.Pool
+var AgentMQPool agtmq.Pool
 
-var CoordinatorMQPool *coormq.Pool
+var CoordinatorMQPool coormq.Pool
 
-var ScannerMQPool *scmq.Pool
+var ScannerMQPool scmq.Pool
 
 func InitMQPool(cfg *stgmq.Config) {
 	AgentMQPool = agtmq.NewPool(cfg)

common/pkgs/cmd/create_rep_package.go

@@ -275,7 +275,7 @@ func pinIPFSFile(nodeID int64, fileHash string) error {
 	if err != nil {
 		return fmt.Errorf("new agent client: %w", err)
 	}
-	defer agtCli.Close()
+	defer stgglb.AgentMQPool.Release(agtCli)
 
 	// 然后让最近节点pin本地上传的文件
 	pinObjResp, err := agtCli.StartPinningObject(agtmq.NewStartPinningObject(fileHash))

common/pkgs/cmd/download_package.go

@@ -38,7 +38,7 @@ func (t *DownloadPackage) Execute(ctx *DownloadPackageContext) error {
 	if err != nil {
 		return fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	getPkgResp, err := coorCli.GetPackage(coormq.NewGetPackage(t.userID, t.packageID))
 	if err != nil {
@@ -65,7 +65,7 @@ func (t *DownloadPackage) downloadRep(ctx *DownloadPackageContext) (iterator.Dow
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	getObjsResp, err := coorCli.GetPackageObjects(coormq.NewGetPackageObjects(t.userID, t.packageID))
 	if err != nil {
@@ -89,7 +89,7 @@ func (t *DownloadPackage) downloadEC(ctx *DownloadPackageContext, pkg model.Pack
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	getObjsResp, err := coorCli.GetPackageObjects(coormq.NewGetPackageObjects(t.userID, t.packageID))
 	if err != nil {

common/pkgs/iterator/ec_object_iterator.go

@@ -48,7 +48,7 @@ func (i *ECObjectIterator) MoveNext() (*IterDownloadingObject, error) {
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	if !i.inited {
 		i.inited = true
@@ -69,7 +69,7 @@ func (i *ECObjectIterator) MoveNext() (*IterDownloadingObject, error) {
 	return item, err
 }
 
-func (iter *ECObjectIterator) doMove(coorCli *coormq.PoolClient) (*IterDownloadingObject, error) {
+func (iter *ECObjectIterator) doMove(coorCli *coormq.Client) (*IterDownloadingObject, error) {
 	obj := iter.objects[iter.currentIndex]
 	ecData := iter.objectECData[iter.currentIndex]
 

common/pkgs/iterator/rep_object_iterator.go

@@ -60,7 +60,7 @@ func (i *RepObjectIterator) MoveNext() (*IterDownloadingObject, error) {
 	if err != nil {
 		return nil, fmt.Errorf("new coordinator client: %w", err)
 	}
-	defer coorCli.Close()
+	defer stgglb.CoordinatorMQPool.Release(coorCli)
 
 	if !i.inited {
 		i.inited = true
@@ -81,7 +81,7 @@ func (i *RepObjectIterator) MoveNext() (*IterDownloadingObject, error) {
 	return item, err
 }
 
-func (i *RepObjectIterator) doMove(coorCli *coormq.PoolClient) (*IterDownloadingObject, error) {
+func (i *RepObjectIterator) doMove(coorCli *coormq.Client) (*IterDownloadingObject, error) {
 	repData := i.objectRepData[i.currentIndex]
 	if len(repData.NodeIDs) == 0 {
 		return nil, fmt.Errorf("no node has this file %s", repData.FileHash)

common/pkgs/mq/agent/client.go

@@ -6,12 +6,12 @@ import (
 )
 
 type Client struct {
-	rabbitCli *mq.RabbitMQClient
+	rabbitCli *mq.RabbitMQTransport
 	id        int64
 }
 
 func NewClient(id int64, cfg *stgmq.Config) (*Client, error) {
-	rabbitCli, err := mq.NewRabbitMQClient(cfg.MakeConnectingURL(), stgmq.MakeAgentQueueName(id), "")
+	rabbitCli, err := mq.NewRabbitMQTransport(cfg.MakeConnectingURL(), stgmq.MakeAgentQueueName(id), "")
 	if err != nil {
 		return nil, err
 	}
@@ -26,36 +26,24 @@ func (c *Client) Close() {
 	c.rabbitCli.Close()
 }
 
-type PoolClient struct {
-	*Client
-	owner *Pool
+type Pool interface {
+	Acquire(id int64) (*Client, error)
+	Release(cli *Client)
 }
 
-func (c *PoolClient) Close() {
-	c.owner.Release(c)
-}
-
-type Pool struct {
+type pool struct {
 	mqcfg *stgmq.Config
 }
 
-func NewPool(mqcfg *stgmq.Config) *Pool {
-	return &Pool{
+func NewPool(mqcfg *stgmq.Config) Pool {
+	return &pool{
 		mqcfg: mqcfg,
 	}
 }
-func (p *Pool) Acquire(id int64) (*PoolClient, error) {
-	cli, err := NewClient(id, p.mqcfg)
-	if err != nil {
-		return nil, err
-	}
-
-	return &PoolClient{
-		Client: cli,
-		owner:  p,
-	}, nil
+func (p *pool) Acquire(id int64) (*Client, error) {
+	return NewClient(id, p.mqcfg)
 }
 
-func (p *Pool) Release(cli *PoolClient) {
-	cli.Client.Close()
+func (p *pool) Release(cli *Client) {
+	cli.Close()
 }

common/pkgs/mq/agent/server.go

@@ -18,8 +18,6 @@ type Service interface {
 type Server struct {
 	service   Service
 	rabbitSvr mq.RabbitMQServer
-
-	OnError func(err error)
 }
 
 func NewServer(svc Service, id int64, cfg *mymq.Config) (*Server, error) {
@@ -51,6 +49,10 @@ func (s *Server) Serve() error {
 	return s.rabbitSvr.Serve()
 }
 
+func (s *Server) OnError(callback func(error)) {
+	s.rabbitSvr.OnError = callback
+}
+
 var msgDispatcher mq.MessageDispatcher = mq.NewMessageDispatcher()
 
 // Register 将Service中的一个接口函数作为指定类型消息的处理函数，同时会注册请求和响应的消息类型

common/pkgs/mq/coordinator/client.go

@@ -6,11 +6,11 @@ import (
 )
 
 type Client struct {
-	rabbitCli *mq.RabbitMQClient
+	rabbitCli *mq.RabbitMQTransport
 }
 
 func NewClient(cfg *stgmq.Config) (*Client, error) {
-	rabbitCli, err := mq.NewRabbitMQClient(cfg.MakeConnectingURL(), stgmq.COORDINATOR_QUEUE_NAME, "")
+	rabbitCli, err := mq.NewRabbitMQTransport(cfg.MakeConnectingURL(), stgmq.COORDINATOR_QUEUE_NAME, "")
 	if err != nil {
 		return nil, err
 	}
@@ -24,36 +24,24 @@ func (c *Client) Close() {
 	c.rabbitCli.Close()
 }
 
-type PoolClient struct {
-	*Client
-	owner *Pool
+type Pool interface {
+	Acquire() (*Client, error)
+	Release(cli *Client)
 }
 
-func (c *PoolClient) Close() {
-	c.owner.Release(c)
-}
-
-type Pool struct {
+type pool struct {
 	mqcfg *stgmq.Config
 }
 
-func NewPool(mqcfg *stgmq.Config) *Pool {
-	return &Pool{
+func NewPool(mqcfg *stgmq.Config) Pool {
+	return &pool{
 		mqcfg: mqcfg,
 	}
 }
-func (p *Pool) Acquire() (*PoolClient, error) {
-	cli, err := NewClient(p.mqcfg)
-	if err != nil {
-		return nil, err
-	}
-
-	return &PoolClient{
-		Client: cli,
-		owner:  p,
-	}, nil
+func (p *pool) Acquire() (*Client, error) {
+	return NewClient(p.mqcfg)
 }
 
-func (p *Pool) Release(cli *PoolClient) {
-	cli.Client.Close()
+func (p *pool) Release(cli *Client) {
+	cli.Close()
 }

common/pkgs/mq/coordinator/server.go

@@ -27,8 +27,6 @@ type Service interface {
 type Server struct {
 	service   Service
 	rabbitSvr mq.RabbitMQServer
-
-	OnError func(err error)
 }
 
 func NewServer(svc Service, cfg *mymq.Config) (*Server, error) {
@@ -59,6 +57,10 @@ func (s *Server) Serve() error {
 	return s.rabbitSvr.Serve()
 }
 
+func (s *Server) OnError(callback func(error)) {
+	s.rabbitSvr.OnError = callback
+}
+
 var msgDispatcher mq.MessageDispatcher = mq.NewMessageDispatcher()
 
 // Register 将Service中的一个接口函数作为指定类型消息的处理函数，同时会注册请求和响应的消息类型

common/pkgs/mq/scanner/client.go

@@ -6,11 +6,11 @@ import (
 )
 
 type Client struct {
-	rabbitCli *mq.RabbitMQClient
+	rabbitCli *mq.RabbitMQTransport
 }
 
 func NewClient(cfg *stgmq.Config) (*Client, error) {
-	rabbitCli, err := mq.NewRabbitMQClient(cfg.MakeConnectingURL(), stgmq.SCANNER_QUEUE_NAME, "")
+	rabbitCli, err := mq.NewRabbitMQTransport(cfg.MakeConnectingURL(), stgmq.SCANNER_QUEUE_NAME, "")
 	if err != nil {
 		return nil, err
 	}
@@ -24,36 +24,24 @@ func (c *Client) Close() {
 	c.rabbitCli.Close()
 }
 
-type PoolClient struct {
-	*Client
-	owner *Pool
+type Pool interface {
+	Acquire() (*Client, error)
+	Release(cli *Client)
 }
 
-func (c *PoolClient) Close() {
-	c.owner.Release(c)
-}
-
-type Pool struct {
+type pool struct {
 	mqcfg *stgmq.Config
 }
 
-func NewPool(mqcfg *stgmq.Config) *Pool {
-	return &Pool{
+func NewPool(mqcfg *stgmq.Config) Pool {
+	return &pool{
 		mqcfg: mqcfg,
 	}
 }
-func (p *Pool) Acquire() (*PoolClient, error) {
-	cli, err := NewClient(p.mqcfg)
-	if err != nil {
-		return nil, err
-	}
-
-	return &PoolClient{
-		Client: cli,
-		owner:  p,
-	}, nil
+func (p *pool) Acquire() (*Client, error) {
+	return NewClient(p.mqcfg)
 }
 
-func (p *Pool) Release(cli *PoolClient) {
-	cli.Client.Close()
+func (p *pool) Release(cli *Client) {
+	cli.Close()
 }

common/pkgs/mq/scanner/event.go

@@ -29,7 +29,3 @@ func NewPostEvent(event scevt.Event, isEmergency bool, dontMerge bool) *PostEven
 func (client *Client) PostEvent(msg *PostEvent) error {
 	return mq.Send(Service.PostEvent, client.rabbitCli, msg)
 }
-
-func init() {
-	mq.RegisterUnionType(scevt.EventTypeUnino)
-}

common/pkgs/mq/scanner/event/agent_check_cache.go

@@ -1,5 +1,7 @@
 package event
 
+var _ = Register[*AgentCheckCache]()
+
 type AgentCheckCache struct {
 	EventBase
 	NodeID     int64    `json:"nodeID"`
@@ -12,7 +14,3 @@ func NewAgentCheckCache(nodeID int64, fileHashes []string) *AgentCheckCache {
 		FileHashes: fileHashes,
 	}
 }
-
-func init() {
-	Register[AgentCheckCache]()
-}

common/pkgs/mq/scanner/event/agent_check_state.go

@@ -1,5 +1,7 @@
 package event
 
+var _ = Register[*AgentCheckState]()
+
 type AgentCheckState struct {
 	EventBase
 	NodeID int64 `json:"nodeID"`
@@ -10,7 +12,3 @@ func NewAgentCheckState(nodeID int64) *AgentCheckState {
 		NodeID: nodeID,
 	}
 }
-
-func init() {
-	Register[AgentCheckState]()
-}

common/pkgs/mq/scanner/event/agent_check_storage.go

@@ -1,5 +1,7 @@
 package event
 
+var _ = Register[*AgentCheckStorage]()
+
 type AgentCheckStorage struct {
 	EventBase
 	StorageID  int64   `json:"storageID"`
@@ -12,7 +14,3 @@ func NewAgentCheckStorage(storageID int64, packageIDs []int64) *AgentCheckStorag
 		PackageIDs: packageIDs,
 	}
 }
-
-func init() {
-	Register[AgentCheckStorage]()
-}

common/pkgs/mq/scanner/event/check_cache.go

@@ -1,5 +1,7 @@
 package event
 
+var _ = Register[*CheckCache]()
+
 type CheckCache struct {
 	EventBase
 	NodeID int64 `json:"nodeID"`
@@ -10,7 +12,3 @@ func NewCheckCache(nodeID int64) *CheckCache {
 		NodeID: nodeID,
 	}
 }
-
-func init() {
-	Register[CheckCache]()
-}

common/pkgs/mq/scanner/event/check_package.go

@@ -1,5 +1,7 @@
 package event
 
+var _ = Register[*CheckPackage]()
+
 type CheckPackage struct {
 	EventBase
 	PackageIDs []int64 `json:"packageIDs"`
@@ -10,7 +12,3 @@ func NewCheckPackage(packageIDs []int64) *CheckPackage {
 		PackageIDs: packageIDs,
 	}
 }
-
-func init() {
-	Register[CheckPackage]()
-}

common/pkgs/mq/scanner/event/check_rep_count.go

@@ -1,5 +1,7 @@
 package event
 
+var _ = Register[*CheckRepCount]()
+
 type CheckRepCount struct {
 	EventBase
 	FileHashes []string `json:"fileHashes"`
@@ -10,7 +12,3 @@ func NewCheckRepCount(fileHashes []string) *CheckRepCount {
 		FileHashes: fileHashes,
 	}
 }
-
-func init() {
-	Register[CheckRepCount]()
-}

common/pkgs/mq/scanner/event/event.go

@@ -1,6 +1,7 @@
 package event
 
 import (
+	"gitlink.org.cn/cloudream/common/pkgs/mq"
 	"gitlink.org.cn/cloudream/common/pkgs/types"
 	myreflect "gitlink.org.cn/cloudream/common/utils/reflect"
 )
@@ -15,6 +16,13 @@ type EventBase struct{}
 
 func (e *EventBase) Noop() {}
 
-func Register[T any]() {
+// 注：此函数必须以var _ = Register[xxx]()的形式被调用，这样才能保证init中RegisterUnionType时
+// TypeUnion不是空的。（因为包级变量初始化比init函数调用先进行）
+func Register[T Event]() any {
 	EventTypeUnino.Add(myreflect.TypeOf[T]())
+	return nil
+}
+
+func init() {
+	mq.RegisterUnionType(EventTypeUnino)
 }

common/pkgs/mq/scanner/server.go

@@ -12,8 +12,6 @@ type Service interface {
 type Server struct {
 	service   Service
 	rabbitSvr mq.RabbitMQServer
-
-	OnError func(err error)
 }
 
 func NewServer(svc Service, cfg *mymq.Config) (*Server, error) {
@@ -45,6 +43,10 @@ func (s *Server) Serve() error {
 	return s.rabbitSvr.Serve()
 }
 
+func (s *Server) OnError(callback func(error)) {
+	s.rabbitSvr.OnError = callback
+}
+
 var msgDispatcher mq.MessageDispatcher = mq.NewMessageDispatcher()
 
 // Register 将Service中的一个接口函数作为指定类型消息的处理函数，同时会注册请求和响应的消息类型

coordinator/main.go

@@ -40,9 +40,9 @@ func main() {
 		logger.Fatalf("new coordinator server failed, err: %s", err.Error())
 	}
 
-	coorSvr.OnError = func(err error) {
+	coorSvr.OnError(func(err error) {
 		logger.Warnf("coordinator server err: %s", err.Error())
-	}
+	})
 
 	// 启动服务
 	go serveCoorServer(coorSvr)

magefiles/main.go

@@ -69,8 +69,13 @@ func Scripts() error {
 	scriptsDir := "./common/assets/scripts"
 
 	info, err := os.Stat(scriptsDir)
-	if errors.Is(err, os.ErrNotExist) || !info.IsDir() {
-		return fmt.Errorf("script directory not exists or is not a directory")
+	if errors.Is(err, os.ErrNotExist) {
+		fmt.Printf("no scripts.\n")
+		return nil
+	}
+
+	if !info.IsDir() {
+		return fmt.Errorf("scripts is not a directory")
 	}
 
 	fullDirPath, err := filepath.Abs(filepath.Join(BuildDir, "scripts"))
@@ -87,8 +92,13 @@ func Confs() error {
 	confDir := "./common/assets/confs"
 
 	info, err := os.Stat(confDir)
-	if errors.Is(err, os.ErrNotExist) || !info.IsDir() {
-		return fmt.Errorf("conf directory not exists or is not a directory")
+	if errors.Is(err, os.ErrNotExist) {
+		fmt.Printf("no confs.\n")
+		return nil
+	}
+
+	if !info.IsDir() {
+		return fmt.Errorf("confs is not a directory")
 	}
 
 	fullDirPath, err := filepath.Abs(filepath.Join(BuildDir, "confs"))
@@ -106,7 +116,7 @@ func Agent() error {
 		OutputName: "agent",
 		OutputDir:  "agent",
 		AssetsDir:  "assets",
-		EntryFile:"agent/main.go",
+		EntryFile:  "agent/main.go",
 	})
 }
 
@@ -115,7 +125,7 @@ func Client() error {
 		OutputName: "client",
 		OutputDir:  "client",
 		AssetsDir:  "assets",
-		EntryFile:"client/main.go",
+		EntryFile:  "client/main.go",
 	})
 }
 
@@ -124,7 +134,7 @@ func Coordinator() error {
 		OutputName: "coordinator",
 		OutputDir:  "coordinator",
 		AssetsDir:  "assets",
-		EntryFile:"coordinator/main.go",
+		EntryFile:  "coordinator/main.go",
 	})
 }
 
@@ -133,6 +143,6 @@ func Scanner() error {
 		OutputName: "scanner",
 		OutputDir:  "scanner",
 		AssetsDir:  "assets",
-		EntryFile:"scanner/main.go",
+		EntryFile:  "scanner/main.go",
 	})
 }

scanner/internal/event/agent_check_cache.go

@@ -126,14 +126,14 @@ func (t *AgentCheckCache) startCheck(execCtx ExecuteContext, isComplete bool, ca
 	log := logger.WithType[AgentCheckCache]("Event")
 
 	// 然后向代理端发送移动文件的请求
-	agentClient, err := stgglb.AgentMQPool.Acquire(t.NodeID)
+	agtCli, err := stgglb.AgentMQPool.Acquire(t.NodeID)
 	if err != nil {
 		log.WithField("NodeID", t.NodeID).Warnf("create agent client failed, err: %s", err.Error())
 		return
 	}
-	defer agentClient.Close()
+	defer stgglb.AgentMQPool.Release(agtCli)
 
-	checkResp, err := agentClient.CheckCache(agtmq.NewCheckCache(isComplete, caches), mq.RequestOption{Timeout: time.Minute})
+	checkResp, err := agtCli.CheckCache(agtmq.NewCheckCache(isComplete, caches), mq.RequestOption{Timeout: time.Minute})
 	if err != nil {
 		log.WithField("NodeID", t.NodeID).Warnf("checking ipfs: %s", err.Error())
 		return
@@ -170,5 +170,5 @@ func (t *AgentCheckCache) startCheck(execCtx ExecuteContext, isComplete bool, ca
 }
 
 func init() {
-	RegisterMessageConvertor(func(msg scevt.AgentCheckCache) Event { return NewAgentCheckCache(msg.NodeID, msg.FileHashes) })
+	RegisterMessageConvertor(func(msg *scevt.AgentCheckCache) Event { return NewAgentCheckCache(msg.NodeID, msg.FileHashes) })
 }

scanner/internal/event/agent_check_state.go

@@ -61,14 +61,14 @@ func (t *AgentCheckState) Execute(execCtx ExecuteContext) {
 		return
 	}
 
-	agentClient, err := stgglb.AgentMQPool.Acquire(t.NodeID)
+	agtCli, err := stgglb.AgentMQPool.Acquire(t.NodeID)
 	if err != nil {
 		log.WithField("NodeID", t.NodeID).Warnf("create agent client failed, err: %s", err.Error())
 		return
 	}
-	defer agentClient.Close()
+	defer stgglb.AgentMQPool.Release(agtCli)
 
-	getResp, err := agentClient.GetState(agtmq.NewGetState(), mq.RequestOption{Timeout: time.Second * 30})
+	getResp, err := agtCli.GetState(agtmq.NewGetState(), mq.RequestOption{Timeout: time.Second * 30})
 	if err != nil {
 		log.WithField("NodeID", t.NodeID).Warnf("getting state: %s", err.Error())
 
@@ -113,5 +113,5 @@ func (t *AgentCheckState) Execute(execCtx ExecuteContext) {
 }
 
 func init() {
-	RegisterMessageConvertor(func(msg scevt.AgentCheckState) Event { return NewAgentCheckState(msg.NodeID) })
+	RegisterMessageConvertor(func(msg *scevt.AgentCheckState) Event { return NewAgentCheckState(msg.NodeID) })
 }

scanner/internal/event/agent_check_storage.go

@@ -143,14 +143,14 @@ func (t *AgentCheckStorage) startCheck(execCtx ExecuteContext, stg model.Storage
 	log := logger.WithType[AgentCheckStorage]("Event")
 
 	// 投递任务
-	agentClient, err := stgglb.AgentMQPool.Acquire(stg.NodeID)
+	agtCli, err := stgglb.AgentMQPool.Acquire(stg.NodeID)
 	if err != nil {
 		log.WithField("NodeID", stg.NodeID).Warnf("create agent client failed, err: %s", err.Error())
 		return
 	}
-	defer agentClient.Close()
+	defer stgglb.AgentMQPool.Release(agtCli)
 
-	checkResp, err := agentClient.StorageCheck(agtmq.NewStorageCheck(stg.StorageID, stg.Directory, isComplete, packages), mq.RequestOption{Timeout: time.Minute})
+	checkResp, err := agtCli.StorageCheck(agtmq.NewStorageCheck(stg.StorageID, stg.Directory, isComplete, packages), mq.RequestOption{Timeout: time.Minute})
 	if err != nil {
 		log.WithField("NodeID", stg.NodeID).Warnf("checking storage: %s", err.Error())
 		return
@@ -195,5 +195,5 @@ func (t *AgentCheckStorage) startCheck(execCtx ExecuteContext, stg model.Storage
 }
 
 func init() {
-	RegisterMessageConvertor(func(msg scevt.AgentCheckStorage) Event { return NewAgentCheckStorage(msg.StorageID, msg.PackageIDs) })
+	RegisterMessageConvertor(func(msg *scevt.AgentCheckStorage) Event { return NewAgentCheckStorage(msg.StorageID, msg.PackageIDs) })
 }

scanner/internal/event/check_cache.go

@@ -80,5 +80,5 @@ func (t *CheckCache) Execute(execCtx ExecuteContext) {
 }
 
 func init() {
-	RegisterMessageConvertor(func(msg scevt.CheckCache) Event { return NewCheckCache(msg.NodeID) })
+	RegisterMessageConvertor(func(msg *scevt.CheckCache) Event { return NewCheckCache(msg.NodeID) })
 }

scanner/internal/event/check_package.go

@@ -53,5 +53,5 @@ func (t *CheckPackage) Execute(execCtx ExecuteContext) {
 }
 
 func init() {
-	RegisterMessageConvertor(func(msg scevt.CheckPackage) Event { return NewCheckPackage(msg.PackageIDs) })
+	RegisterMessageConvertor(func(msg *scevt.CheckPackage) Event { return NewCheckPackage(msg.PackageIDs) })
 }

scanner/internal/event/check_rep_count.go

@@ -211,5 +211,5 @@ func chooseDeleteAvaiRepNodes(allNodes []model.Node, curAvaiRepNodes []model.Nod
 }
 
 func init() {
-	RegisterMessageConvertor(func(msg scevt.CheckRepCount) Event { return NewCheckRepCount(msg.FileHashes) })
+	RegisterMessageConvertor(func(msg *scevt.CheckRepCount) Event { return NewCheckRepCount(msg.FileHashes) })
 }

scanner/internal/event/event.go

@@ -36,7 +36,7 @@ var msgDispatcher = typedispatcher.NewTypeDispatcher[Event]()
 func FromMessage(msg scevt.Event) (Event, error) {
 	event, ok := msgDispatcher.Dispatch(msg)
 	if !ok {
-		return nil, fmt.Errorf("unknow event message type: %s", reflect.TypeOf(msg).Name())
+		return nil, fmt.Errorf("unknow event message type: %s", reflect.TypeOf(msg).String())
 	}
 
 	return event, nil

scanner/main.go

@@ -53,9 +53,10 @@ func main() {
 	if err != nil {
 		logger.Fatalf("new agent server failed, err: %s", err.Error())
 	}
-	agtSvr.OnError = func(err error) {
+	agtSvr.OnError(func(err error) {
 		logger.Warnf("agent server err: %s", err.Error())
-	}
+	})
+
 	go serveScannerServer(agtSvr, &wg)
 
 	tickExecutor := tickevent.NewExecutor(tickevent.ExecuteArgs{