调整项目结构

go.mod

@@ -6,11 +6,8 @@ require (
 	github.com/ipfs/go-ipfs-api v0.6.0
 	github.com/samber/lo v1.38.1
 	gitlink.org.cn/cloudream/common v0.0.0
-	gitlink.org.cn/cloudream/ec v0.0.0
-	gitlink.org.cn/cloudream/proto v0.0.0
-	gitlink.org.cn/cloudream/rabbitmq v0.0.0
-	google.golang.org/grpc v1.54.0
-	magefiles v0.0.0
+	gitlink.org.cn/cloudream/storage-common v0.0.0
+	google.golang.org/grpc v1.57.0
 )
 
 require (
@@ -36,7 +33,6 @@ require (
 	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
 	github.com/libp2p/go-flow-metrics v0.1.0 // indirect
 	github.com/libp2p/go-libp2p v0.27.0 // indirect
-	github.com/magefile/mage v1.15.0 // indirect
 	github.com/minio/sha256-simd v1.0.0 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
@@ -51,13 +47,11 @@ require (
 	github.com/multiformats/go-multihash v0.2.1 // indirect
 	github.com/multiformats/go-multistream v0.4.1 // indirect
 	github.com/multiformats/go-varint v0.0.7 // indirect
-	github.com/otiai10/copy v1.12.0 // indirect
 	github.com/sirupsen/logrus v1.9.2 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
-	github.com/streadway/amqp v1.0.0 // indirect
+	github.com/streadway/amqp v1.1.0 // indirect
 	github.com/whyrusleeping/tar-utils v0.0.0-20201201191210-20a61371de5b // indirect
 	github.com/zyedidia/generic v1.2.1 // indirect
-	gitlink.org.cn/cloudream/db v0.0.0 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.9 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.9 // indirect
 	go.etcd.io/etcd/client/v3 v3.5.9 // indirect
@@ -70,22 +64,15 @@ require (
 	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.7.0 // indirect
 	golang.org/x/text v0.9.0 // indirect
-	google.golang.org/genproto v0.0.0-20230403163135-c38d8f061ccd // indirect
+	google.golang.org/genproto v0.0.0-20230526161137-0005af68ea54 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20230525234035-dd9d682886f9 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 // indirect
 	google.golang.org/protobuf v1.30.0 // indirect
 	lukechampine.com/blake3 v1.1.7 // indirect
 )
 
-go 1.18
-
 // 运行go mod tidy时需要将下面几行取消注释
-// replace gitlink.org.cn/cloudream/ec => ../ec
-// 
-// replace gitlink.org.cn/cloudream/proto => ../proto
-// 
-// replace gitlink.org.cn/cloudream/rabbitmq => ../rabbitmq
-// 
-// replace gitlink.org.cn/cloudream/common => ../common
-// 
-// replace gitlink.org.cn/cloudream/db => ../db
-// 
-// replace magefiles => ../magefiles
+
+replace gitlink.org.cn/cloudream/common => ../../common
+
+replace gitlink.org.cn/cloudream/storage-common => ../storage-common

go.sum

@@ -62,8 +62,6 @@ github.com/libp2p/go-flow-metrics v0.1.0 h1:0iPhMI8PskQwzh57jB9WxIuIOQ0r+15PChFG
 github.com/libp2p/go-flow-metrics v0.1.0/go.mod h1:4Xi8MX8wj5aWNDAZttg6UPmc0ZrnFNsMtpsYUClFtro=
 github.com/libp2p/go-libp2p v0.27.0 h1:QbhrTuB0ln9j9op6yAOR0o+cx/qa9NyNZ5ov0Tql8ZU=
 github.com/libp2p/go-libp2p v0.27.0/go.mod h1:FAvvfQa/YOShUYdiSS03IR9OXzkcJXwcNA2FUCh9ImE=
-github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=
-github.com/magefile/mage v1.15.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
 github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g=
 github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM=
 github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
@@ -92,9 +90,6 @@ github.com/multiformats/go-multistream v0.4.1 h1:rFy0Iiyn3YT0asivDUIR05leAdwZq3d
 github.com/multiformats/go-multistream v0.4.1/go.mod h1:Mz5eykRVAjJWckE2U78c6xqdtyNUEhKSM0Lwar2p77Q=
 github.com/multiformats/go-varint v0.0.7 h1:sWSGR+f/eu5ABZA2ZpYKBILXTTs9JWpdEM/nEGOHFS8=
 github.com/multiformats/go-varint v0.0.7/go.mod h1:r8PUYw/fD/SjBCiKOoDlGF6QawOELpZAu9eioSos/OU=
-github.com/otiai10/copy v1.12.0 h1:cLMgSQnXBs1eehF0Wy/FAGsgDTDmAqFR7rQylBb1nDY=
-github.com/otiai10/copy v1.12.0/go.mod h1:rSaLseMUsZFFbsFGc7wCJnnkTAvdc5L6VWxPE4308Ww=
-github.com/otiai10/mint v1.5.1 h1:XaPLeE+9vGbuyEHem1JNk3bYc7KKqyI/na0/mLd/Kks=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
@@ -106,8 +101,8 @@ github.com/smartystreets/assertions v1.13.1 h1:Ef7KhSmjZcK6AVf9YbJdvPYG9avaF0Zxu
 github.com/smartystreets/goconvey v1.8.0 h1:Oi49ha/2MURE0WexF052Z0m+BNSGirfjg5RL+JXWq3w=
 github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
 github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
-github.com/streadway/amqp v1.0.0 h1:kuuDrUJFZL1QYL9hUNuCxNObNzB0bV/ZG5jV3RWAQgo=
-github.com/streadway/amqp v1.0.0/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
+github.com/streadway/amqp v1.1.0 h1:py12iX8XSyI7aN/3dUT8DFIDJazNJsVJdxNVEpnQTZM=
+github.com/streadway/amqp v1.1.0/go.mod h1:WYSrTEYHOXHd0nwFeUXAe2G2hRnQT+deZJJf88uS9Bg=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
@@ -175,10 +170,14 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/genproto v0.0.0-20230403163135-c38d8f061ccd h1:sLpv7bNL1AsX3fdnWh9WVh7ejIzXdOc1RRHGeAmeStU=
-google.golang.org/genproto v0.0.0-20230403163135-c38d8f061ccd/go.mod h1:UUQDJDOlWu4KYeJZffbWgBkS1YFobzKbLVfK69pe0Ak=
-google.golang.org/grpc v1.54.0 h1:EhTqbhiYeixwWQtAEZAxmV9MGqcjEU2mFx52xCzNyag=
-google.golang.org/grpc v1.54.0/go.mod h1:PUSEXI6iWghWaB6lXM4knEgpJNu2qUcKfDtNci3EC2g=
+google.golang.org/genproto v0.0.0-20230526161137-0005af68ea54 h1:9NWlQfY2ePejTmfwUH1OWwmznFa+0kKcHGPDvcPza9M=
+google.golang.org/genproto v0.0.0-20230526161137-0005af68ea54/go.mod h1:zqTuNwFlFRsw5zIts5VnzLQxSRqh+CGOTVMlYbY0Eyk=
+google.golang.org/genproto/googleapis/api v0.0.0-20230525234035-dd9d682886f9 h1:m8v1xLLLzMe1m5P+gCTF8nJB9epwZQUBERm20Oy1poQ=
+google.golang.org/genproto/googleapis/api v0.0.0-20230525234035-dd9d682886f9/go.mod h1:vHYtlOoi6TsQ3Uk2yxR7NI5z8uoV+3pZtR4jmHIkRig=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 h1:0nDDozoAU19Qb2HwhXadU8OcsiO/09cnTqhUtq2MEOM=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA=
+google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw=
+google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=

internal/config/config.go

@@ -5,7 +5,7 @@ import (
 	log "gitlink.org.cn/cloudream/common/pkg/logger"
 	c "gitlink.org.cn/cloudream/common/utils/config"
 	"gitlink.org.cn/cloudream/common/utils/ipfs"
-	racfg "gitlink.org.cn/cloudream/rabbitmq/config"
+	racfg "gitlink.org.cn/cloudream/storage-common/pkgs/mq/config"
 )
 
 type Config struct {

internal/services/cmd/agent.go

@@ -2,11 +2,11 @@ package cmd
 
 import (
 	"gitlink.org.cn/cloudream/common/consts"
-	ramsg "gitlink.org.cn/cloudream/rabbitmq/message"
-	agtmsg "gitlink.org.cn/cloudream/rabbitmq/message/agent"
+	"gitlink.org.cn/cloudream/common/pkg/mq"
+	agtmsg "gitlink.org.cn/cloudream/storage-common/pkgs/mq/message/agent"
 )
 
-func (svc *Service) GetState(msg *agtmsg.GetState) (*agtmsg.GetStateResp, *ramsg.CodeMessage) {
+func (svc *Service) GetState(msg *agtmsg.GetState) (*agtmsg.GetStateResp, *mq.CodeMessage) {
 	var ipfsState string
 
 	if svc.ipfs.IsUp() {
@@ -15,5 +15,5 @@ func (svc *Service) GetState(msg *agtmsg.GetState) (*agtmsg.GetStateResp, *ramsg
 		ipfsState = consts.IPFSStateOK
 	}
 
-	return ramsg.ReplyOK(agtmsg.NewGetStateRespBody(ipfsState))
+	return mq.ReplyOK(agtmsg.NewGetStateRespBody(ipfsState))
 }

internal/services/cmd/ipfs.go

@@ -7,17 +7,17 @@ import (
 	"gitlink.org.cn/cloudream/common/consts"
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
 	"gitlink.org.cn/cloudream/common/pkg/logger"
-	ramsg "gitlink.org.cn/cloudream/rabbitmq/message"
-	agtmsg "gitlink.org.cn/cloudream/rabbitmq/message/agent"
+	"gitlink.org.cn/cloudream/common/pkg/mq"
 	"gitlink.org.cn/cloudream/storage-agent/internal/config"
 	"gitlink.org.cn/cloudream/storage-agent/internal/task"
+	agtmsg "gitlink.org.cn/cloudream/storage-common/pkgs/mq/message/agent"
 )
 
-func (svc *Service) CheckIPFS(msg *agtmsg.CheckIPFS) (*agtmsg.CheckIPFSResp, *ramsg.CodeMessage) {
+func (svc *Service) CheckIPFS(msg *agtmsg.CheckIPFS) (*agtmsg.CheckIPFSResp, *mq.CodeMessage) {
 	filesMap, err := svc.ipfs.GetPinnedFiles()
 	if err != nil {
 		logger.Warnf("get pinned files from ipfs failed, err: %s", err.Error())
-		return ramsg.ReplyFailed[agtmsg.CheckIPFSResp](errorcode.OperationFailed, "get pinned files from ipfs failed")
+		return mq.ReplyFailed[agtmsg.CheckIPFSResp](errorcode.OperationFailed, "get pinned files from ipfs failed")
 	}
 
 	// TODO 根据锁定清单过滤被锁定的文件的记录
@@ -28,7 +28,7 @@ func (svc *Service) CheckIPFS(msg *agtmsg.CheckIPFS) (*agtmsg.CheckIPFSResp, *ra
 	}
 }
 
-func (svc *Service) checkIncrement(msg *agtmsg.CheckIPFS, filesMap map[string]shell.PinInfo) (*agtmsg.CheckIPFSResp, *ramsg.CodeMessage) {
+func (svc *Service) checkIncrement(msg *agtmsg.CheckIPFS, filesMap map[string]shell.PinInfo) (*agtmsg.CheckIPFSResp, *mq.CodeMessage) {
 	var entries []agtmsg.CheckIPFSRespEntry
 	for _, cache := range msg.Caches {
 		_, ok := filesMap[cache.FileHash]
@@ -60,10 +60,10 @@ func (svc *Service) checkIncrement(msg *agtmsg.CheckIPFS, filesMap map[string]sh
 
 	// 增量情况下，不需要对filesMap中没检查的记录进行处理
 
-	return ramsg.ReplyOK(agtmsg.NewCheckIPFSResp(entries))
+	return mq.ReplyOK(agtmsg.NewCheckIPFSResp(entries))
 }
 
-func (svc *Service) checkComplete(msg *agtmsg.CheckIPFS, filesMap map[string]shell.PinInfo) (*agtmsg.CheckIPFSResp, *ramsg.CodeMessage) {
+func (svc *Service) checkComplete(msg *agtmsg.CheckIPFS, filesMap map[string]shell.PinInfo) (*agtmsg.CheckIPFSResp, *mq.CodeMessage) {
 	var entries []agtmsg.CheckIPFSRespEntry
 	for _, cache := range msg.Caches {
 		_, ok := filesMap[cache.FileHash]
@@ -103,5 +103,5 @@ func (svc *Service) checkComplete(msg *agtmsg.CheckIPFS, filesMap map[string]she
 		entries = append(entries, agtmsg.NewCheckIPFSRespEntry(hash, agtmsg.CHECK_IPFS_RESP_OP_CREATE_TEMP))
 	}
 
-	return ramsg.ReplyOK(agtmsg.NewCheckIPFSResp(entries))
+	return mq.ReplyOK(agtmsg.NewCheckIPFSResp(entries))
 }

internal/services/cmd/object.go

@@ -5,12 +5,12 @@ import (
 
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
 	log "gitlink.org.cn/cloudream/common/pkg/logger"
-	ramsg "gitlink.org.cn/cloudream/rabbitmq/message"
-	agtmsg "gitlink.org.cn/cloudream/rabbitmq/message/agent"
+	"gitlink.org.cn/cloudream/common/pkg/mq"
 	"gitlink.org.cn/cloudream/storage-agent/internal/task"
+	agtmsg "gitlink.org.cn/cloudream/storage-common/pkgs/mq/message/agent"
 )
 
-func (svc *Service) StartPinningObject(msg *agtmsg.StartPinningObject) (*agtmsg.StartPinningObjectResp, *ramsg.CodeMessage) {
+func (svc *Service) StartPinningObject(msg *agtmsg.StartPinningObject) (*agtmsg.StartPinningObjectResp, *mq.CodeMessage) {
 	log.WithField("FileHash", msg.FileHash).Debugf("pin object")
 
 	tsk := svc.taskManager.StartComparable(task.NewIPFSPin(msg.FileHash))
@@ -18,18 +18,18 @@ func (svc *Service) StartPinningObject(msg *agtmsg.StartPinningObject) (*agtmsg.
 	if tsk.Error() != nil {
 		log.WithField("FileHash", msg.FileHash).
 			Warnf("pin object failed, err: %s", tsk.Error().Error())
-		return ramsg.ReplyFailed[agtmsg.StartPinningObjectResp](errorcode.OperationFailed, "pin object failed")
+		return mq.ReplyFailed[agtmsg.StartPinningObjectResp](errorcode.OperationFailed, "pin object failed")
 	}
 
-	return ramsg.ReplyOK(agtmsg.NewStartPinningObjectResp(tsk.ID()))
+	return mq.ReplyOK(agtmsg.NewStartPinningObjectResp(tsk.ID()))
 }
 
-func (svc *Service) WaitPinningObject(msg *agtmsg.WaitPinningObject) (*agtmsg.WaitPinningObjectResp, *ramsg.CodeMessage) {
+func (svc *Service) WaitPinningObject(msg *agtmsg.WaitPinningObject) (*agtmsg.WaitPinningObjectResp, *mq.CodeMessage) {
 	log.WithField("TaskID", msg.TaskID).Debugf("wait pinning object")
 
 	tsk := svc.taskManager.FindByID(msg.TaskID)
 	if tsk == nil {
-		return ramsg.ReplyFailed[agtmsg.WaitPinningObjectResp](errorcode.TaskNotFound, "task not found")
+		return mq.ReplyFailed[agtmsg.WaitPinningObjectResp](errorcode.TaskNotFound, "task not found")
 	}
 
 	if msg.WaitTimeoutMs == 0 {
@@ -40,7 +40,7 @@ func (svc *Service) WaitPinningObject(msg *agtmsg.WaitPinningObject) (*agtmsg.Wa
 			errMsg = tsk.Error().Error()
 		}
 
-		return ramsg.ReplyOK(agtmsg.NewWaitPinningObjectResp(true, errMsg))
+		return mq.ReplyOK(agtmsg.NewWaitPinningObjectResp(true, errMsg))
 
 	} else {
 		if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs)) {
@@ -50,9 +50,9 @@ func (svc *Service) WaitPinningObject(msg *agtmsg.WaitPinningObject) (*agtmsg.Wa
 				errMsg = tsk.Error().Error()
 			}
 
-			return ramsg.ReplyOK(agtmsg.NewWaitPinningObjectResp(true, errMsg))
+			return mq.ReplyOK(agtmsg.NewWaitPinningObjectResp(true, errMsg))
 		}
 
-		return ramsg.ReplyOK(agtmsg.NewWaitPinningObjectResp(false, ""))
+		return mq.ReplyOK(agtmsg.NewWaitPinningObjectResp(false, ""))
 	}
 }

internal/services/cmd/storage.go

@@ -10,19 +10,19 @@ import (
 
 	"github.com/samber/lo"
 	"gitlink.org.cn/cloudream/common/consts"
-	"gitlink.org.cn/cloudream/common/models"
 	"gitlink.org.cn/cloudream/common/pkg/logger"
 	"gitlink.org.cn/cloudream/common/utils"
-	"gitlink.org.cn/cloudream/ec"
 	"gitlink.org.cn/cloudream/storage-agent/internal/config"
 	"gitlink.org.cn/cloudream/storage-agent/internal/task"
+	"gitlink.org.cn/cloudream/storage-common/models"
+	"gitlink.org.cn/cloudream/storage-common/pkgs/ec"
 
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
-	ramsg "gitlink.org.cn/cloudream/rabbitmq/message"
-	agtmsg "gitlink.org.cn/cloudream/rabbitmq/message/agent"
+	"gitlink.org.cn/cloudream/common/pkg/mq"
+	agtmsg "gitlink.org.cn/cloudream/storage-common/pkgs/mq/message/agent"
 )
 
-func (service *Service) StartStorageMoveObject(msg *agtmsg.StartStorageMoveObject) (*agtmsg.StartStorageMoveObjectResp, *ramsg.CodeMessage) {
+func (service *Service) StartStorageMoveObject(msg *agtmsg.StartStorageMoveObject) (*agtmsg.StartStorageMoveObjectResp, *mq.CodeMessage) {
 	// TODO  修改文件名，可用objectname
 	outFileName := utils.MakeMoveOperationFileName(msg.ObjectID, msg.UserID)
 	objectDir := filepath.Dir(msg.ObjectName)
@@ -32,22 +32,22 @@ func (service *Service) StartStorageMoveObject(msg *agtmsg.StartStorageMoveObjec
 		taskID, err := service.moveRepObject(repRed, outFilePath)
 		if err != nil {
 			logger.Warnf("move rep object as %s failed, err: %s", outFilePath, err.Error())
-			return ramsg.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "move rep object failed")
+			return mq.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "move rep object failed")
 		}
 
-		return ramsg.ReplyOK(agtmsg.NewStartStorageMoveObjectResp(taskID))
+		return mq.ReplyOK(agtmsg.NewStartStorageMoveObjectResp(taskID))
 
 	} else if repRed, ok := msg.Redundancy.(models.ECRedundancyData); ok {
 		taskID, err := service.moveEcObject(msg.ObjectID, msg.FileSize, repRed, outFilePath)
 		if err != nil {
 			logger.Warnf("move ec object as %s failed, err: %s", outFilePath, err.Error())
-			return ramsg.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "move ec object failed")
+			return mq.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "move ec object failed")
 		}
 
-		return ramsg.ReplyOK(agtmsg.NewStartStorageMoveObjectResp(taskID))
+		return mq.ReplyOK(agtmsg.NewStartStorageMoveObjectResp(taskID))
 	}
 
-	return ramsg.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "not rep or ec object???")
+	return mq.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "not rep or ec object???")
 }
 
 func (svc *Service) moveRepObject(repData models.RepRedundancyData, outFilePath string) (string, error) {
@@ -67,12 +67,12 @@ func (svc *Service) moveEcObject(objID int64, fileSize int64, ecData models.ECRe
 	return tsk.ID(), nil
 }
 
-func (svc *Service) WaitStorageMoveObject(msg *agtmsg.WaitStorageMoveObject) (*agtmsg.WaitStorageMoveObjectResp, *ramsg.CodeMessage) {
+func (svc *Service) WaitStorageMoveObject(msg *agtmsg.WaitStorageMoveObject) (*agtmsg.WaitStorageMoveObjectResp, *mq.CodeMessage) {
 	logger.WithField("TaskID", msg.TaskID).Debugf("wait moving object")
 
 	tsk := svc.taskManager.FindByID(msg.TaskID)
 	if tsk == nil {
-		return ramsg.ReplyFailed[agtmsg.WaitStorageMoveObjectResp](errorcode.TaskNotFound, "task not found")
+		return mq.ReplyFailed[agtmsg.WaitStorageMoveObjectResp](errorcode.TaskNotFound, "task not found")
 	}
 
 	if msg.WaitTimeoutMs == 0 {
@@ -83,7 +83,7 @@ func (svc *Service) WaitStorageMoveObject(msg *agtmsg.WaitStorageMoveObject) (*a
 			errMsg = tsk.Error().Error()
 		}
 
-		return ramsg.ReplyOK(agtmsg.NewWaitStorageMoveObjectResp(true, errMsg))
+		return mq.ReplyOK(agtmsg.NewWaitStorageMoveObjectResp(true, errMsg))
 
 	} else {
 		if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs)) {
@@ -93,20 +93,20 @@ func (svc *Service) WaitStorageMoveObject(msg *agtmsg.WaitStorageMoveObject) (*a
 				errMsg = tsk.Error().Error()
 			}
 
-			return ramsg.ReplyOK(agtmsg.NewWaitStorageMoveObjectResp(true, errMsg))
+			return mq.ReplyOK(agtmsg.NewWaitStorageMoveObjectResp(true, errMsg))
 		}
 
-		return ramsg.ReplyOK(agtmsg.NewWaitStorageMoveObjectResp(false, ""))
+		return mq.ReplyOK(agtmsg.NewWaitStorageMoveObjectResp(false, ""))
 	}
 }
 
-func (svc *Service) StorageCheck(msg *agtmsg.StorageCheck) (*agtmsg.StorageCheckResp, *ramsg.CodeMessage) {
+func (svc *Service) StorageCheck(msg *agtmsg.StorageCheck) (*agtmsg.StorageCheckResp, *mq.CodeMessage) {
 	dirFullPath := filepath.Join(config.Cfg().StorageBaseDir, msg.Directory)
 
 	infos, err := ioutil.ReadDir(dirFullPath)
 	if err != nil {
 		logger.Warnf("list storage directory failed, err: %s", err.Error())
-		return ramsg.ReplyOK(agtmsg.NewStorageCheckResp(
+		return mq.ReplyOK(agtmsg.NewStorageCheckResp(
 			err.Error(),
 			nil,
 		))
@@ -121,7 +121,7 @@ func (svc *Service) StorageCheck(msg *agtmsg.StorageCheck) (*agtmsg.StorageCheck
 	}
 }
 
-func (svc *Service) checkStorageIncrement(msg *agtmsg.StorageCheck, fileInfos []fs.FileInfo) (*agtmsg.StorageCheckResp, *ramsg.CodeMessage) {
+func (svc *Service) checkStorageIncrement(msg *agtmsg.StorageCheck, fileInfos []fs.FileInfo) (*agtmsg.StorageCheckResp, *mq.CodeMessage) {
 	infosMap := make(map[string]fs.FileInfo)
 	for _, info := range fileInfos {
 		infosMap[info.Name()] = info
@@ -145,10 +145,10 @@ func (svc *Service) checkStorageIncrement(msg *agtmsg.StorageCheck, fileInfos []
 
 	// 增量情况下，不需要对infosMap中没检查的记录进行处理
 
-	return ramsg.ReplyOK(agtmsg.NewStorageCheckResp(consts.StorageDirectoryStateOK, entries))
+	return mq.ReplyOK(agtmsg.NewStorageCheckResp(consts.StorageDirectoryStateOK, entries))
 }
 
-func (svc *Service) checkStorageComplete(msg *agtmsg.StorageCheck, fileInfos []fs.FileInfo) (*agtmsg.StorageCheckResp, *ramsg.CodeMessage) {
+func (svc *Service) checkStorageComplete(msg *agtmsg.StorageCheck, fileInfos []fs.FileInfo) (*agtmsg.StorageCheckResp, *mq.CodeMessage) {
 
 	infosMap := make(map[string]fs.FileInfo)
 	for _, info := range fileInfos {
@@ -171,7 +171,7 @@ func (svc *Service) checkStorageComplete(msg *agtmsg.StorageCheck, fileInfos []f
 		}
 	}
 
-	return ramsg.ReplyOK(agtmsg.NewStorageCheckResp(consts.StorageDirectoryStateOK, entries))
+	return mq.ReplyOK(agtmsg.NewStorageCheckResp(consts.StorageDirectoryStateOK, entries))
 }
 
 func decode(inBufs []chan []byte, outBufs []chan []byte, blockSeq []int, ecK int, numPacket int64) {
@@ -213,20 +213,20 @@ func decode(inBufs []chan []byte, outBufs []chan []byte, blockSeq []int, ecK int
 	}
 }
 
-func (svc *Service) StartStorageUploadRepObject(msg *agtmsg.StartStorageUploadRepObject) (*agtmsg.StartStorageUploadRepObjectResp, *ramsg.CodeMessage) {
+func (svc *Service) StartStorageUploadRepObject(msg *agtmsg.StartStorageUploadRepObject) (*agtmsg.StartStorageUploadRepObjectResp, *mq.CodeMessage) {
 	fullPath := filepath.Join(config.Cfg().StorageBaseDir, msg.StorageDirectory, msg.FilePath)
 
 	file, err := os.Open(fullPath)
 	if err != nil {
 		logger.Warnf("opening file %s: %s", fullPath, err.Error())
-		return nil, ramsg.Failed(errorcode.OperationFailed, "open file failed")
+		return nil, mq.Failed(errorcode.OperationFailed, "open file failed")
 	}
 
 	fileInfo, err := file.Stat()
 	if err != nil {
 		file.Close()
 		logger.Warnf("getting file %s state: %s", fullPath, err.Error())
-		return nil, ramsg.Failed(errorcode.OperationFailed, "get file info failed")
+		return nil, mq.Failed(errorcode.OperationFailed, "get file info failed")
 	}
 	fileSize := fileInfo.Size()
 
@@ -239,19 +239,19 @@ func (svc *Service) StartStorageUploadRepObject(msg *agtmsg.StartStorageUploadRe
 
 	// Task会关闭文件流
 	tsk := svc.taskManager.StartNew(task.NewUploadRepObjects(msg.UserID, msg.BucketID, uploadObjects, msg.RepCount))
-	return ramsg.ReplyOK(agtmsg.NewStartStorageUploadRepObjectResp(tsk.ID()))
+	return mq.ReplyOK(agtmsg.NewStartStorageUploadRepObjectResp(tsk.ID()))
 }
 
-func (svc *Service) WaitStorageUploadRepObject(msg *agtmsg.WaitStorageUploadRepObject) (*agtmsg.WaitStorageUploadRepObjectResp, *ramsg.CodeMessage) {
+func (svc *Service) WaitStorageUploadRepObject(msg *agtmsg.WaitStorageUploadRepObject) (*agtmsg.WaitStorageUploadRepObjectResp, *mq.CodeMessage) {
 	tsk := svc.taskManager.FindByID(msg.TaskID)
 	if tsk == nil {
-		return nil, ramsg.Failed(errorcode.TaskNotFound, "task not found")
+		return nil, mq.Failed(errorcode.TaskNotFound, "task not found")
 	}
 
 	if msg.WaitTimeoutMs == 0 {
 		tsk.Wait()
 	} else if !tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs)) {
-		return ramsg.ReplyOK(agtmsg.NewWaitStorageUploadRepObjectResp(false, "", 0, ""))
+		return mq.ReplyOK(agtmsg.NewWaitStorageUploadRepObjectResp(false, "", 0, ""))
 	}
 
 	uploadTask := tsk.Body().(*task.UploadRepObjects)
@@ -266,5 +266,5 @@ func (svc *Service) WaitStorageUploadRepObject(msg *agtmsg.WaitStorageUploadRepO
 		errMsg = uploadRet.Error.Error()
 	}
 
-	return ramsg.ReplyOK(agtmsg.NewWaitStorageUploadRepObjectResp(true, errMsg, uploadRet.ObjectID, uploadRet.FileHash))
+	return mq.ReplyOK(agtmsg.NewWaitStorageUploadRepObjectResp(true, errMsg, uploadRet.ObjectID, uploadRet.FileHash))
 }

internal/services/grpc/grpc_service.go

@@ -7,7 +7,7 @@ import (
 	log "gitlink.org.cn/cloudream/common/pkg/logger"
 	myio "gitlink.org.cn/cloudream/common/utils/io"
 	"gitlink.org.cn/cloudream/common/utils/ipfs"
-	agentserver "gitlink.org.cn/cloudream/proto"
+	agentserver "gitlink.org.cn/cloudream/storage-common/pkgs/proto"
 )
 
 type GRPCService struct {

internal/task/ec_read.go

@@ -7,10 +7,10 @@ import (
 	"path/filepath"
 	"time"
 
-	"gitlink.org.cn/cloudream/common/models"
 	"gitlink.org.cn/cloudream/common/pkg/logger"
-	"gitlink.org.cn/cloudream/ec"
 	"gitlink.org.cn/cloudream/storage-agent/internal/config"
+	"gitlink.org.cn/cloudream/storage-common/models"
+	"gitlink.org.cn/cloudream/storage-common/pkgs/ec"
 )
 
 type EcRead struct {

internal/task/task.go

@@ -4,7 +4,7 @@ import (
 	distsvc "gitlink.org.cn/cloudream/common/pkg/distlock/service"
 	"gitlink.org.cn/cloudream/common/pkg/task"
 	"gitlink.org.cn/cloudream/common/utils/ipfs"
-	coorcli "gitlink.org.cn/cloudream/rabbitmq/client/coordinator"
+	coorcli "gitlink.org.cn/cloudream/storage-common/pkgs/mq/client/coordinator"
 )
 
 type TaskContext struct {

internal/task/upload_rep_objects.go

@@ -10,13 +10,13 @@ import (
 	"gitlink.org.cn/cloudream/common/pkg/distlock/reqbuilder"
 	"gitlink.org.cn/cloudream/common/pkg/logger"
 	"gitlink.org.cn/cloudream/common/utils"
-	mygrpc "gitlink.org.cn/cloudream/common/utils/grpc"
 	"gitlink.org.cn/cloudream/common/utils/ipfs"
 	"gitlink.org.cn/cloudream/storage-agent/internal/config"
+	mygrpc "gitlink.org.cn/cloudream/storage-common/utils/grpc"
 
-	agentcaller "gitlink.org.cn/cloudream/proto"
-	ramsg "gitlink.org.cn/cloudream/rabbitmq/message"
-	coormsg "gitlink.org.cn/cloudream/rabbitmq/message/coordinator"
+	ramsg "gitlink.org.cn/cloudream/storage-common/pkgs/mq/message"
+	coormsg "gitlink.org.cn/cloudream/storage-common/pkgs/mq/message/coordinator"
+	agentcaller "gitlink.org.cn/cloudream/storage-common/pkgs/proto"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"

main.go

@@ -9,14 +9,14 @@ import (
 	distsvc "gitlink.org.cn/cloudream/common/pkg/distlock/service"
 	log "gitlink.org.cn/cloudream/common/pkg/logger"
 	"gitlink.org.cn/cloudream/common/utils/ipfs"
-	agentserver "gitlink.org.cn/cloudream/proto"
 	"gitlink.org.cn/cloudream/storage-agent/internal/config"
 	"gitlink.org.cn/cloudream/storage-agent/internal/task"
+	agentserver "gitlink.org.cn/cloudream/storage-common/pkgs/proto"
 
 	"google.golang.org/grpc"
 
-	"gitlink.org.cn/cloudream/rabbitmq/client/coordinator"
-	rasvr "gitlink.org.cn/cloudream/rabbitmq/server/agent"
+	"gitlink.org.cn/cloudream/storage-common/pkgs/mq/client/coordinator"
+	rasvr "gitlink.org.cn/cloudream/storage-common/pkgs/mq/server/agent"
 
 	cmdsvc "gitlink.org.cn/cloudream/storage-agent/internal/services/cmd"
 	grpcsvc "gitlink.org.cn/cloudream/storage-agent/internal/services/grpc"

status_report.go

@@ -7,9 +7,9 @@ import (
 	"gitlink.org.cn/cloudream/common/consts"
 	log "gitlink.org.cn/cloudream/common/pkg/logger"
 	"gitlink.org.cn/cloudream/common/utils"
-	coorcli "gitlink.org.cn/cloudream/rabbitmq/client/coordinator"
-	coormsg "gitlink.org.cn/cloudream/rabbitmq/message/coordinator"
 	"gitlink.org.cn/cloudream/storage-agent/internal/config"
+	coorcli "gitlink.org.cn/cloudream/storage-common/pkgs/mq/client/coordinator"
+	coormsg "gitlink.org.cn/cloudream/storage-common/pkgs/mq/message/coordinator"
 )
 
 func reportStatus(wg *sync.WaitGroup) {