feature: implement the dfgetTaskMgr

Signed-off-by: Starnop <[email protected]>

Author: starnop

apis/swagger.yml

@@ -827,6 +827,12 @@ definitions:
         description: |
             The status of Dfget download process.
         enum: ["WAITING", "RUNNING", "FAILED", "SUCCESS",]
+      peerID:
+        type: "string"
+        description: |
+          PeerID uniquely identifies a peer, and the cID uniquely identifies a 
+          download task belonging to a peer. One peer can initiate multiple download tasks, 
+          which means that one peer corresponds to multiple cIDs.
  
   ErrorResponse:
     type: "object"

apis/types/df_get_task.go

@@ -2,32 +2,143 @@ package mgr
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/dragonflyoss/Dragonfly/apis/types"
+	cutil "github.com/dragonflyoss/Dragonfly/common/util"
+	errorType "github.com/dragonflyoss/Dragonfly/supernode/errors"
+
+	"github.com/pkg/errors"
 )
 
 // DfgetTaskMgr as an interface defines all operations against DfgetTask.
 // A DfgetTask represents a download process initiated by dfget or other clients.
 type DfgetTaskMgr interface {
-	// Add a new dfgetTask, we use clientID to identify a dfgetTask uniquely.
+	// Add a new dfgetTask, we use clientID and taskID to identify a dfgetTask uniquely.
 	// ClientID should be generated by dfget, supernode will use it directly.
 	// NOTE: We should create a new dfgetTask for each download process,
 	//       even if the downloads initiated by the same machine.
 	Add(ctx context.Context, dfgetTask *types.DfGetTask) error
 
-	// Get get a dfgetTask info with specified clientID.
-	Get(ctx context.Context, clientID string) (dfgetTask *types.DfGetTask, err error)
+	// Get a dfgetTask info with specified clientID and taskID.
+	Get(ctx context.Context, clientID, taskID string) (dfgetTask *types.DfGetTask, err error)
 
-	// List return the list of dfgetTask.
+	// List returns the list of dfgetTask.
 	List(ctx context.Context, filter map[string]string) (dfgetTaskList []*types.DfGetTask, err error)
 
-	// Delete a dfgetTask with clientID.
-	Delete(ctx context.Context, clientID string) error
+	// Delete a dfgetTask with clientID and taskID.
+	Delete(ctx context.Context, clientID, taskID string) error
 
-	// UpdateStatus update the status of dfgetTask with specified clientID.
+	// UpdateStatus update the status of dfgetTask with specified clientID and taskID.
 	// Supernode will update the status of dfgetTask in the following situations:
 	// 1. after init the dfgetTask
 	// 2. when success/fail to download some pieces
 	// 3. when the entire download process ends in success or failure
-	UpdateStatus(ctx context.Context, clientID string, status string) error
+	UpdateStatus(ctx context.Context, clientID, taskID, status string) error
+}
+
+// DfgetTaskManager is an implementation of the interface of DfgetTaskMgr.
+type DfgetTaskManager struct {
+	dfgetTaskStore *Store
+}
+
+// NewDfgetTaskManager returns a new DfgetTaskManager.
+func NewDfgetTaskManager() (*DfgetTaskManager, error) {
+	return &DfgetTaskManager{
+		dfgetTaskStore: NewStore(),
+	}, nil
+}
+
+// Add a new dfgetTask, we use clientID and taskID to identify a dfgetTask uniquely.
+// ClientID should be generated by dfget, supernode will use it directly.
+// NOTE: We should create a new dfgetTask for each download process,
+//       even if the downloads initiated by the same machine.
+func (dtm *DfgetTaskManager) Add(ctx context.Context, dfgetTask *types.DfGetTask) error {
+	if cutil.IsEmptyStr(dfgetTask.Path) {
+		return errors.Wrapf(errorType.ErrEmptyValue, "Path")
+	}
+
+	if cutil.IsEmptyStr(dfgetTask.PeerID) {
+		return errors.Wrapf(errorType.ErrEmptyValue, "PeerID")
+	}
+
+	key, err := generateKey(dfgetTask.CID, dfgetTask.TaskID)
+	if err != nil {
+		return err
+	}
+
+	// the default status of DfgetTask is WAITING
+	if cutil.IsEmptyStr(dfgetTask.Status) {
+		dfgetTask.Status = types.DfGetTaskStatusWAITING
+	}
+
+	// TODO: should we verify that the peerID is valid here.
+
+	dtm.dfgetTaskStore.Put(key, dfgetTask)
+	return nil
+}
+
+// Get a dfgetTask info with specified clientID and taskID.
+func (dtm *DfgetTaskManager) Get(ctx context.Context, clientID, taskID string) (dfgetTask *types.DfGetTask, err error) {
+	return dtm.getDfgetTask(clientID, taskID)
+}
+
+// List returns the list of dfgetTask.
+func (dtm *DfgetTaskManager) List(ctx context.Context, filter map[string]string) (dfgetTaskList []*types.DfGetTask, err error) {
+	return nil, nil
+}
+
+// Delete a dfgetTask with clientID and taskID.
+func (dtm *DfgetTaskManager) Delete(ctx context.Context, clientID, taskID string) error {
+	key, err := generateKey(clientID, taskID)
+	if err != nil {
+		return err
+	}
+
+	return dtm.dfgetTaskStore.Delete(key)
+}
+
+// UpdateStatus update the status of dfgetTask with specified clientID and taskID.
+func (dtm *DfgetTaskManager) UpdateStatus(ctx context.Context, clientID, taskID, status string) error {
+	dfgetTask, err := dtm.getDfgetTask(clientID, taskID)
+	if err != nil {
+		return err
+	}
+
+	if dfgetTask.Status != types.DfGetTaskStatusSUCCESS {
+		dfgetTask.Status = status
+	}
+
+	return nil
+}
+
+// getDfgetTask gets a DfGetTask from dfgetTaskStore with specified clientID and taskID.
+func (dtm *DfgetTaskManager) getDfgetTask(clientID, taskID string) (*types.DfGetTask, error) {
+	key, err := generateKey(clientID, taskID)
+	if err != nil {
+		return nil, err
+	}
+
+	v, err := dtm.dfgetTaskStore.Get(key)
+	if err != nil {
+		return nil, err
+	}
+
+	if dfgetTask, ok := v.(*types.DfGetTask); ok {
+		return dfgetTask, nil
+	}
+	return nil, errors.Wrapf(errorType.ErrConvertFailed, "clientID: %s, taskID: %s: %v", clientID, taskID, v)
+}
+
+// generateKey gengerates a key for a dfgetTask.
+func generateKey(cID, taskID string) (string, error) {
+	if cutil.IsEmptyStr(cID) {
+		return "", errors.Wrapf(errorType.ErrEmptyValue, "cID")
+	}
+
+	if cutil.IsEmptyStr(taskID) {
+		return "", errors.Wrapf(errorType.ErrEmptyValue, "taskID")
+	}
+
+	return fmt.Sprintf("%s%s%s", cID, "@", taskID), nil
 }

supernode/daemon/mgr/dfget_task_mgr.go

@@ -0,0 +1,85 @@
+/*
+ * Copyright The Dragonfly Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package mgr
+
+import (
+	"context"
+
+	"github.com/dragonflyoss/Dragonfly/apis/types"
+	"github.com/dragonflyoss/Dragonfly/supernode/errors"
+
+	"github.com/go-check/check"
+)
+
+func init() {
+	check.Suite(&DfgetTaskMgrTestSuite{})
+}
+
+type DfgetTaskMgrTestSuite struct {
+}
+
+func (s *DfgetTaskMgrTestSuite) TestDfgetTaskMgr(c *check.C) {
+	dfgetTaskManager, _ := NewDfgetTaskManager()
+	clientID := "foo"
+	taskID := "00c4e7b174af7ed61c414b36ef82810ac0c98142c03e5748c00e1d1113f3c882"
+
+	// Add
+	dfgetTask := &types.DfGetTask{
+		CID:       clientID,
+		Path:      "/peer/file/taskFileName",
+		PieceSize: 4 * 1024 * 1024,
+		TaskID:    taskID,
+		PeerID:    "foo-192.168.10.11-1553838710990554281",
+	}
+
+	err := dfgetTaskManager.Add(context.Background(), dfgetTask)
+	c.Check(err, check.IsNil)
+
+	// Get
+	dt, err := dfgetTaskManager.Get(context.Background(), clientID, taskID)
+	c.Check(err, check.IsNil)
+	c.Check(dt, check.DeepEquals, &types.DfGetTask{
+		CID:       clientID,
+		Path:      "/peer/file/taskFileName",
+		PieceSize: 4 * 1024 * 1024,
+		TaskID:    taskID,
+		Status:    types.DfGetTaskStatusWAITING,
+		PeerID:    "foo-192.168.10.11-1553838710990554281",
+	})
+
+	// UpdateStatus
+	err = dfgetTaskManager.UpdateStatus(context.Background(), clientID, taskID, types.DfGetTaskStatusSUCCESS)
+	c.Check(err, check.IsNil)
+
+	dt, err = dfgetTaskManager.Get(context.Background(), clientID, taskID)
+	c.Check(err, check.IsNil)
+	c.Check(dt, check.DeepEquals, &types.DfGetTask{
+		CID:       clientID,
+		Path:      "/peer/file/taskFileName",
+		PieceSize: 4 * 1024 * 1024,
+		TaskID:    taskID,
+		Status:    types.DfGetTaskStatusSUCCESS,
+		PeerID:    "foo-192.168.10.11-1553838710990554281",
+	})
+
+	// Delete
+	err = dfgetTaskManager.Delete(context.Background(), clientID, taskID)
+	c.Check(err, check.IsNil)
+
+	dt, err = dfgetTaskManager.Get(context.Background(), clientID, taskID)
+	c.Check(errors.IsDataNotFound(err), check.Equals, true)
+}