66 lines
2.7 KiB
Go
66 lines
2.7 KiB
Go
package services
|
|
|
|
import (
|
|
"time"
|
|
|
|
"gitlink.org.cn/cloudream/common/pkgs/mq"
|
|
dctrlmq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/dcontroller"
|
|
schtsk "gitlink.org.cn/cloudream/scheduler/executor/internal/task"
|
|
)
|
|
|
|
func (svc *Service) StartStorageLoadPackage(msg *dctrlmq.StartStorageLoadPackage) (*dctrlmq.StartStorageLoadPackageResp, *mq.CodeMessage) {
|
|
tsk := svc.taskManager.StartNew(schtsk.NewStorageLoadPackage(msg.UserID, msg.PackageID, msg.StorageID))
|
|
return mq.ReplyOK(dctrlmq.NewStartStorageLoadPackageResp(tsk.ID()))
|
|
}
|
|
|
|
func (svc *Service) WaitStorageLoadPackage(msg *dctrlmq.WaitStorageLoadPackage) (*dctrlmq.WaitStorageLoadPackageResp, *mq.CodeMessage) {
|
|
tsk := svc.taskManager.FindByID(msg.TaskID)
|
|
if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs) * time.Millisecond) {
|
|
errMsg := ""
|
|
if tsk.Error() != nil {
|
|
errMsg = tsk.Error().Error()
|
|
}
|
|
|
|
return mq.ReplyOK(dctrlmq.NewWaitStorageLoadPackageResp(true, errMsg))
|
|
}
|
|
return mq.ReplyOK(dctrlmq.NewWaitStorageLoadPackageResp(false, ""))
|
|
}
|
|
|
|
func (svc *Service) StartStorageCreatePackage(msg *dctrlmq.StartStorageCreatePackage) (*dctrlmq.StartStorageCreatePackageResp, *mq.CodeMessage) {
|
|
tsk := svc.taskManager.StartNew(schtsk.NewStorageCreatePackage(msg.UserID, msg.StorageID, msg.Path, msg.BucketID, msg.Name, msg.Redundancy))
|
|
return mq.ReplyOK(dctrlmq.NewStartStorageCreatePackageResp(tsk.ID()))
|
|
}
|
|
|
|
func (svc *Service) WaitStorageCreatePackage(msg *dctrlmq.WaitStorageCreatePackage) (*dctrlmq.WaitStorageCreatePackageResp, *mq.CodeMessage) {
|
|
tsk := svc.taskManager.FindByID(msg.TaskID)
|
|
if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs) * time.Millisecond) {
|
|
tskBody := tsk.Body().(*schtsk.StorageCreatePackage)
|
|
|
|
errMsg := ""
|
|
if tsk.Error() != nil {
|
|
errMsg = tsk.Error().Error()
|
|
}
|
|
|
|
return mq.ReplyOK(dctrlmq.NewWaitStorageCreatePackageResp(true, errMsg, tskBody.Result.PackageID))
|
|
}
|
|
return mq.ReplyOK(dctrlmq.NewWaitStorageCreatePackageResp(false, "", 0))
|
|
}
|
|
|
|
func (svc *Service) StartCacheMovePackage(msg *dctrlmq.StartCacheMovePackage) (*dctrlmq.StartCacheMovePackageResp, *mq.CodeMessage) {
|
|
tsk := svc.taskManager.StartNew(schtsk.NewCacheMovePackage(msg.UserID, msg.PackageID, msg.NodeID))
|
|
return mq.ReplyOK(dctrlmq.NewStartCacheMovePackageResp(tsk.ID()))
|
|
}
|
|
|
|
func (svc *Service) WaitCacheMovePackage(msg *dctrlmq.WaitCacheMovePackage) (*dctrlmq.WaitCacheMovePackageResp, *mq.CodeMessage) {
|
|
tsk := svc.taskManager.FindByID(msg.TaskID)
|
|
if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs) * time.Millisecond) {
|
|
errMsg := ""
|
|
if tsk.Error() != nil {
|
|
errMsg = tsk.Error().Error()
|
|
}
|
|
|
|
return mq.ReplyOK(dctrlmq.NewWaitCacheMovePackageResp(true, errMsg))
|
|
}
|
|
return mq.ReplyOK(dctrlmq.NewWaitCacheMovePackageResp(false, ""))
|
|
}
|