Removes ReqId field in PartialData

Author: grussorusso

internal/test/partial_data_test.go

@@ -14,9 +14,8 @@ func TestPartialDataMarshaling(t *testing.T) {
 	data["num"] = 2
 	data["list"] = []string{"uno", "due", "tre"}
 	partialData := workflow.PartialData{
-		ReqId:   workflow.ReqId("abc"),
-		ForTask: "fai13p102",
-		Data:    data,
+		Task: "fai13p102",
+		Data: data,
 	}
 	marshal, errMarshal := json.Marshal(partialData)
 	u.AssertNilMsg(t, errMarshal, "error during marshaling")

internal/test/workflow_integration_test.go

@@ -513,11 +513,11 @@ func TestResumeWorkflow(t *testing.T) {
 	request := workflow.NewRequest(shortuuid.New(), wflow, params)
 
 	progress := workflow.InitProgress(workflow.ReqId(request.Id), wflow)
-	pd := workflow.NewPartialData(workflow.ReqId(request.Id), wflow.Start.Id, request.Params)
+	pd := workflow.NewPartialData(wflow.Start.Id, request.Params)
 
 	err = workflow.SaveProgress(progress)
 	u.AssertNil(t, err)
-	err = workflow.SavePartialData(pd)
+	err = workflow.SavePartialData(pd, workflow.ReqId(request.Id))
 	u.AssertNil(t, err)
 
 	resumedRequest := workflow.NewRequest(request.Id, wflow, params)

internal/workflow/partial_data.go

@@ -13,9 +13,8 @@ import (
 
 // PartialData is saved separately from progressData to avoid cluttering the Progress struct and each Serverledge node's cache
 type PartialData struct {
-	ReqId   ReqId  // request referring to this partial data
-	ForTask TaskId // task that should receive this partial data
-	Data    map[string]interface{}
+	Task TaskId
+	Data map[string]interface{}
 }
 
 func (pd PartialData) Equals(pd2 *PartialData) bool {
@@ -33,30 +32,28 @@ func (pd PartialData) Equals(pd2 *PartialData) bool {
 		}
 	}
 
-	return pd.ReqId == pd2.ReqId && pd.ForTask == pd2.ForTask
+	return pd.Task == pd2.Task
 }
 
 func (pd PartialData) String() string {
 	return fmt.Sprintf(`PartialData{
-		Id:    %s,
-		ForTask:  %s,
+		Task:  %s,
 		Data:     %v,
-	}`, pd.ReqId, pd.ForTask, pd.Data)
+	}`, pd.Task, pd.Data)
 }
 
-func NewPartialData(reqId ReqId, forTask TaskId, data map[string]interface{}) *PartialData {
+func NewPartialData(task TaskId, data map[string]interface{}) *PartialData {
 	return &PartialData{
-		ReqId:   reqId,
-		ForTask: forTask,
-		Data:    data,
+		Task: task,
+		Data: data,
 	}
 }
 
 func getPartialDataEtcdKey(reqId ReqId, nodeId TaskId) string {
 	return fmt.Sprintf("/partialData/%s/%s", reqId, nodeId)
 }
 
-func SavePartialData(pd *PartialData) error {
+func SavePartialData(pd *PartialData, reqId ReqId) error {
 	cli, err := utils.GetEtcdClient()
 	if err != nil {
 		return err
@@ -68,7 +65,7 @@ func SavePartialData(pd *PartialData) error {
 		return fmt.Errorf("could not marshal progress: %v", err)
 	}
 	// saves the json object into etcd
-	key := getPartialDataEtcdKey(pd.ReqId, pd.ForTask)
+	key := getPartialDataEtcdKey(reqId, pd.Task)
 	log.Printf("Saving PD on etcd with key: %s\n", key)
 
 	_, err = cli.Put(ctx, key, string(payload))

internal/workflow/workflow.go

@@ -214,7 +214,7 @@ func (workflow *Workflow) Execute(r *Request, input *PartialData, progress *Prog
 			progress.Complete(task.GetId())
 
 			nextTask := task.GetNext()
-			outputData = NewPartialData(ReqId(r.Id), nextTask, output)
+			outputData = NewPartialData(nextTask, output)
 			if workflow.IsTaskEligibleForExecution(nextTask, progress) {
 				progress.ReadyToExecute = append(progress.ReadyToExecute, nextTask)
 			}
@@ -245,7 +245,7 @@ func (workflow *Workflow) Execute(r *Request, input *PartialData, progress *Prog
 			}
 			progress.Complete(task.GetId())
 
-			outputData = NewPartialData(ReqId(r.Id), nextTaskId, input.Data)
+			outputData = NewPartialData(nextTaskId, input.Data)
 			if workflow.IsTaskEligibleForExecution(nextTaskId, progress) {
 				progress.ReadyToExecute = append(progress.ReadyToExecute, nextTaskId)
 			}
@@ -262,7 +262,7 @@ func (workflow *Workflow) Execute(r *Request, input *PartialData, progress *Prog
 		if err != nil {
 			return nil, progress, false, err
 		}
-		err = SavePartialData(input)
+		err = SavePartialData(input, ReqId(r.Id))
 		if err != nil {
 			return nil, progress, false, err
 		}
@@ -398,7 +398,7 @@ func (workflow *Workflow) Invoke(r *Request) error {
 
 	if !r.Resuming {
 		progress = InitProgress(requestId, workflow)
-		pd = NewPartialData(requestId, workflow.Start.Id, r.Params)
+		pd = NewPartialData(workflow.Start.Id, r.Params)
 	} else {
 		progress, err = RetrieveProgress(requestId)
 		if err != nil {
@@ -430,7 +430,7 @@ func (workflow *Workflow) Invoke(r *Request) error {
 			if err != nil {
 				return fmt.Errorf("Could not save progress: %v", err)
 			}
-			err = SavePartialData(pd)
+			err = SavePartialData(pd, ReqId(r.Id))
 			if err != nil {
 				return fmt.Errorf("Could not save partial data: %v", err)
 			}