task.go

package tq

import (
	"encoding/json"
	"log"
)

type Task struct {

	// The message ID that gets generated by the broker.
	// This field is only added when an event gets consumed.
	Id string `json:"id"`

	// The key of the task function that will execute the job.
	Key string `json:"key"`

	// Holds the serialized signature.
	Payload []byte `json:"payload"`

	// Maximum retry count for task execution to succeed.
	RetryCount int `json:"retry_count,string"`

	// Time duration between each retry attempt.
	RetryTimeout int `json:"retry_timeout,string"`

	// If task processing doesn't complete within the timeout, the task will be retried
	Timeout int64 `json:"timeout,string"`

    // Encode the result into the task.
    // This allows us to only with a single Stream interface, rather than a TaskStream and ResultStream.
    Result
}

func (s Task) Encode() map[string]interface{} {

	var values map[string]interface{}

	data, err := json.Marshal(s)
	if err != nil {
		log.Fatalf("Unable to marshall task: %s", err)
	}

	err = json.Unmarshal(data, &values)
	if err != nil {
		panic(err)
	}

	return values
}

func DecodeTask(msgId string, val map[string]interface{}) *Task {

	jsonbody, err := json.Marshal(val)
	if err != nil {
		panic(err)
	}

	msg := Task{}
	if err := json.Unmarshal(jsonbody, &msg); err != nil {
		panic(err)
	}

	msg.Id = msgId

	return &msg
}