diff --git a/client/internal/http/presigned.go b/client/internal/http/presigned.go index 926ef36..3e4d73d 100644 --- a/client/internal/http/presigned.go +++ b/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") diff --git a/client/internal/http/server.go b/client/internal/http/server.go index 9726f25..66faf64 100644 --- a/client/internal/http/server.go +++ b/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) diff --git a/common/models/datamap.go b/common/models/datamap.go index 5b3b47a..2629472 100644 --- a/common/models/datamap.go +++ b/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 { diff --git a/common/pkgs/ioswitch2/ops2/range.go b/common/pkgs/ioswitch2/ops2/range.go index 4bc70fa..78e99dd 100644 --- a/common/pkgs/ioswitch2/ops2/range.go +++ b/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() diff --git a/common/pkgs/ioswitch2/ops2/segment.go b/common/pkgs/ioswitch2/ops2/segment.go index f1ab0b8..525e330 100644 --- a/common/pkgs/ioswitch2/ops2/segment.go +++ b/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 } } diff --git a/common/pkgs/ioswitchlrc/ops2/range.go b/common/pkgs/ioswitchlrc/ops2/range.go index 4bc70fa..78e99dd 100644 --- a/common/pkgs/ioswitchlrc/ops2/range.go +++ b/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() diff --git a/coordinator/internal/cmd/migrate.go b/coordinator/internal/cmd/migrate.go index 309b832..a73b856 100644 --- a/coordinator/internal/cmd/migrate.go +++ b/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{}) diff --git a/coordinator/internal/mq/package.go b/coordinator/internal/mq/package.go index 999234b..f285d26 100644 --- a/coordinator/internal/mq/package.go +++ b/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))