func writeResults(jobID string, task *Task, ttl time.Duration, rows *sql.Rows, jobber *Jobber) (int64, error) {
	var numRows int64
	name, backend := task.ResultBackends.GetRandom()
	jobber.Logger.Printf("sending results form '%s' to '%s'", jobID, name)
	w, err := backend.NewResultSet(jobID, task.Name, ttl)
	if err != nil {
		return numRows, fmt.Errorf("error writing columns to result backend: %v", err)
	}
	defer w.Close()
	cols, err := rows.Columns()
	if err != nil {
		return numRows, err
	}
	numCols := len(cols)
	if !w.IsColTypesRegistered() {
		colTypes, err := rows.ColumnTypes()
		if err != nil {
			return numRows, err
		}
		w.RegisterColTypes(cols, colTypes)
	}
	if err := w.WriteCols(cols); err != nil {
		return numRows, fmt.Errorf("error writing columns to result backend: %v", err)
	}
	var (
		resCols     = make([]interface{}, numCols)
		resPointers = make([]interface{}, numCols)
	)
	for i := 0; i < numCols; i++ {
		resPointers[i] = &resCols[i]
	}
	for rows.Next() {
		if err := rows.Scan(resPointers...); err != nil {
			return numRows, err
		}
		if err := w.WriteRow(resCols); err != nil {
			return numRows, fmt.Errorf("error writing row to result backend: %v", err)
		}
		numRows++
	}
	if err := w.Flush(); err != nil {
		return numRows, fmt.Errorf("error flushing results to result backend: %v", err)
	}
	return numRows, nil
}
writeResults writes results from an SQL query to a result backend.
package main
import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sync"
	"time"
	machinery "github.com/RichardKnop/machinery/v1"
	"github.com/RichardKnop/machinery/v1/config"
	"github.com/RichardKnop/machinery/v1/tasks"
	"github.com/knadh/sql-jobber/models"
	uuid "github.com/satori/go.uuid"
)
var (
	jobContexts = make(map[string]context.CancelFunc)
	jobMutex    = sync.RWMutex{}
)
func createJobSignature(j models.JobReq, taskName string, ttl int, jobber *Jobber) (tasks.Signature, error) {
	task, ok := jobber.Tasks[taskName]
	if !ok {
		return tasks.Signature{}, fmt.Errorf("unrecognized task: %s", taskName)
	}
	if s, err := jobber.Machinery.GetBackend().GetState(j.JobID); err == nil && !s.IsCompleted() {
		return tasks.Signature{}, fmt.Errorf("job '%s' is already running", j.JobID)
	}
	if j.JobID == "" {
		j.JobID = fmt.Sprintf("job_%v", uuid.NewV4())
	}
	args := append([]tasks.Arg{
		{Type: "string", Value: j.JobID},
		{Type: "string", Value: taskName},
		{Type: "int", Value: ttl},
	}, sliceToTaskArgs(j.Args)...)
	var eta *time.Time
	if j.ETA != "" {
		e, err := time.Parse("2006-01-02 15:04:05", j.ETA)
		if err != nil {
			return tasks.Signature{}, fmt.Errorf("error parsing ETA: %v", err)
		}
		eta = &e
	}
	if j.Queue == "" {
		j.Queue = task.Queue
	}
	return tasks.Signature{
		Name:       taskName,
		UUID:       j.JobID,
		RoutingKey: j.Queue,
		RetryCount: j.Retries,
		Args:       args,
		ETA:        eta,
	}, nil
}
func executeTask(jobID, taskName string, ttl time.Duration, args []interface{}, task *Task, jobber *Jobber) (int64, error) {
	if _, err := jobber.Machinery.GetBackend().GetState(jobID); err != nil {
		return 0, errors.New("the job was canceled")
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer func() {
		cancel()
		jobMutex.Lock()
		delete(jobContexts, jobID)
		jobMutex.Unlock()
	}()
	jobMutex.Lock()
	jobContexts[jobID] = cancel
	jobMutex.Unlock()
	var (
		rows *sql.Rows
		err  error
	)
	if task.Stmt != nil {
		rows, err = task.Stmt.QueryContext(ctx, args...)
	} else {
		_, db := task.DBs.GetRandom()
		rows, err = db.QueryContext(ctx, task.Raw, args...)
	}
	if err != nil {
		if err == context.Canceled {
			return 0, errors.New("the job was canceled")
		}
		return 0, fmt.Errorf("task SQL query execution failed: %v", err)
	}
	defer rows.Close()
	return writeResults(jobID, task, ttl, rows, jobber)
}