Merge branch 'master' into feature_gxh

client/internal/http/presigned.go

@@ -27,6 +27,26 @@ func (s *Server) Presigned() *PresignedService {
 	}
 }
 
+func (s *PresignedService) ObjectListByPath(ctx *gin.Context) {
+	log := logger.WithField("HTTP", "Presigned.ObjectListByPath")
+
+	var req cdsapi.PresignedObjectListByPath
+	if err := ctx.ShouldBindQuery(&req); err != nil {
+		log.Warnf("binding body: %s", err.Error())
+		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
+		return
+	}
+
+	resp, err := s.svc.ObjectSvc().GetByPath(req.ObjectListByPath)
+	if err != nil {
+		log.Warnf("listing objects: %s", err.Error())
+		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, fmt.Sprintf("listing objects: %v", err)))
+		return
+	}
+
+	ctx.JSON(http.StatusOK, OK(resp))
+}
+
 func (s *PresignedService) ObjectDownloadByPath(ctx *gin.Context) {
 	log := logger.WithField("HTTP", "Presigned.ObjectDownloadByPath")
 

client/internal/http/server.go

@@ -124,6 +124,7 @@ func (s *Server) routeV1(eg *gin.Engine, rt gin.IRoutes) {
 	rt.POST(cdsapi.ObjectUploadPartPath, s.Object().UploadPart)
 	rt.POST(cdsapi.ObjectCompleteMultipartUploadPath, s.Object().CompleteMultipartUpload)
 
+	rt.GET(cdsapi.PresignedObjectListByPathPath, s.awsAuth.PresignedAuth, s.Presigned().ObjectListByPath)
 	rt.GET(cdsapi.PresignedObjectDownloadByPathPath, s.awsAuth.PresignedAuth, s.Presigned().ObjectDownloadByPath)
 	rt.GET(cdsapi.PresignedObjectDownloadPath, s.awsAuth.PresignedAuth, s.Presigned().ObjectDownload)
 	rt.POST(cdsapi.PresignedObjectUploadPath, s.awsAuth.PresignedAuth, s.Presigned().ObjectUpload)

common/models/datamap.go

@@ -458,9 +458,11 @@ func (b *BodyNewPackage) OnUnionSerializing() {
 // Package克隆的事件
 type BodyPackageCloned struct {
 	serder.Metadata `union:"PackageCloned"`
-	Type            string           `json:"type"`
-	SourcePackageID cdssdk.PackageID `json:"sourcePackageID"`
-	NewPackage      cdssdk.Package   `json:"newPackage"`
+	Type            string            `json:"type"`
+	SourcePackageID cdssdk.PackageID  `json:"sourcePackageID"`
+	NewPackage      cdssdk.Package    `json:"newPackage"`
+	SourceObjectIDs []cdssdk.ObjectID `json:"sourceObjectIDs"` // 原本的ObjectID
+	NewObjectIDs    []cdssdk.ObjectID `json:"newObjectIDs"`    // 复制后的新ObjectID，与SourceObjectIDs一一对应
 }
 
 func (b *BodyPackageCloned) GetBodyType() string {

common/pkgs/ioswitch2/ops2/range.go

@@ -35,6 +35,7 @@ func (o *Range) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
 	for o.Offset > 0 {
 		rdCnt := math2.Min(o.Offset, int64(len(buf)))
 		rd, err := input.Stream.Read(buf[:rdCnt])
+		o.Offset -= int64(rd)
 		if err == io.EOF {
 			// 输入流不够长度也不报错，只是产生一个空的流
 			break
@@ -42,7 +43,6 @@ func (o *Range) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
 		if err != nil {
 			return err
 		}
-		o.Offset -= int64(rd)
 	}
 
 	fut := future.NewSetVoid()

common/pkgs/ioswitch2/ops2/segment.go

@@ -86,12 +86,20 @@ func (o *SegmentJoin) Read(buf []byte) (int, error) {
 		}
 
 		n, err := o.nextStream.Read(buf)
-		if err == io.EOF {
-			o.nextStream.Close()
-			o.nextStream = nil
+		if err == nil {
+			return n, err
+		}
+
+		if err != io.EOF {
+			return n, err
+		}
+
+		o.nextStream.Close()
+		o.nextStream = nil
+		if n == 0 {
 			continue
 		}
-		return n, err
+		return n, nil
 	}
 }
 

common/pkgs/ioswitchlrc/ops2/range.go

@@ -35,6 +35,7 @@ func (o *Range) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
 	for o.Offset > 0 {
 		rdCnt := math2.Min(o.Offset, int64(len(buf)))
 		rd, err := input.Stream.Read(buf[:rdCnt])
+		o.Offset -= int64(rd)
 		if err == io.EOF {
 			// 输入流不够长度也不报错，只是产生一个空的流
 			break
@@ -42,7 +43,6 @@ func (o *Range) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
 		if err != nil {
 			return err
 		}
-		o.Offset -= int64(rd)
 	}
 
 	fut := future.NewSetVoid()

coordinator/internal/cmd/migrate.go

@@ -40,6 +40,7 @@ func migrate(configPath string) {
 		fmt.Println(err)
 		os.Exit(1)
 	}
+	db = db.Set("gorm:table_options", "CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci")
 
 	migrateOne(db, cdssdk.Bucket{})
 	migrateOne(db, model.Cache{})

coordinator/internal/mq/package.go

@@ -160,6 +160,8 @@ func (svc *Service) DeletePackage(msg *coormq.DeletePackage) (*coormq.DeletePack
 
 func (svc *Service) ClonePackage(msg *coormq.ClonePackage) (*coormq.ClonePackageResp, *mq.CodeMessage) {
 	var pkg cdssdk.Package
+	var oldObjIDs []cdssdk.ObjectID
+	var newObjIDs []cdssdk.ObjectID
 	err := svc.db2.DoTx(func(tx db2.SQLContext) error {
 		var err error
 
@@ -198,6 +200,9 @@ func (svc *Service) ClonePackage(msg *coormq.ClonePackage) (*coormq.ClonePackage
 		oldToNew := make(map[cdssdk.ObjectID]cdssdk.ObjectID)
 		for i, obj := range clonedObjs {
 			oldToNew[objs[i].ObjectID] = obj.ObjectID
+
+			oldObjIDs = append(oldObjIDs, objs[i].ObjectID)
+			newObjIDs = append(newObjIDs, obj.ObjectID)
 		}
 
 		clonedBlks := make([]stgmod.ObjectBlock, len(objBlks))
@@ -224,6 +229,8 @@ func (svc *Service) ClonePackage(msg *coormq.ClonePackage) (*coormq.ClonePackage
 	svc.evtPub.Publish(&stgmod.BodyPackageCloned{
 		SourcePackageID: msg.PackageID,
 		NewPackage:      pkg,
+		SourceObjectIDs: oldObjIDs,
+		NewObjectIDs:    newObjIDs,
 	})
 
 	return mq.ReplyOK(coormq.RespClonePackage(pkg))