Support not using backends on a per task level (#36)

Author: cameron-dunn-sublime

v1/backends/redis/redis.go

@@ -7,15 +7,14 @@ import (
 	"sync"
 	"time"
 
-	"github.com/go-redsync/redsync/v4"
-	redsyncredis "github.com/go-redsync/redsync/v4/redis/redigo"
-	"github.com/gomodule/redigo/redis"
-
 	"github.com/RichardKnop/machinery/v1/backends/iface"
 	"github.com/RichardKnop/machinery/v1/common"
 	"github.com/RichardKnop/machinery/v1/config"
 	"github.com/RichardKnop/machinery/v1/log"
 	"github.com/RichardKnop/machinery/v1/tasks"
+	"github.com/go-redsync/redsync/v4"
+	redsyncredis "github.com/go-redsync/redsync/v4/redis/redigo"
+	"github.com/gomodule/redigo/redis"
 )
 
 // Backend represents a Redis result backend
@@ -158,6 +157,10 @@ func (b *Backend) mergeNewTaskState(conn redis.Conn, newState *tasks.TaskState)
 
 // SetStatePending updates task state to PENDING
 func (b *Backend) SetStatePending(signature *tasks.Signature) error {
+	if signature.NoBackend {
+		return nil
+	}
+
 	conn := b.open()
 	defer conn.Close()
 
@@ -167,6 +170,10 @@ func (b *Backend) SetStatePending(signature *tasks.Signature) error {
 
 // SetStateReceived updates task state to RECEIVED
 func (b *Backend) SetStateReceived(signature *tasks.Signature) error {
+	if signature.NoBackend {
+		return nil
+	}
+
 	conn := b.open()
 	defer conn.Close()
 
@@ -177,6 +184,10 @@ func (b *Backend) SetStateReceived(signature *tasks.Signature) error {
 
 // SetStateStarted updates task state to STARTED
 func (b *Backend) SetStateStarted(signature *tasks.Signature) error {
+	if signature.NoBackend {
+		return nil
+	}
+
 	conn := b.open()
 	defer conn.Close()
 
@@ -187,6 +198,10 @@ func (b *Backend) SetStateStarted(signature *tasks.Signature) error {
 
 // SetStateRetry updates task state to RETRY
 func (b *Backend) SetStateRetry(signature *tasks.Signature) error {
+	if signature.NoBackend {
+		return nil
+	}
+
 	conn := b.open()
 	defer conn.Close()
 
@@ -197,6 +212,10 @@ func (b *Backend) SetStateRetry(signature *tasks.Signature) error {
 
 // SetStateSuccess updates task state to SUCCESS
 func (b *Backend) SetStateSuccess(signature *tasks.Signature, results []*tasks.TaskResult) error {
+	if signature.NoBackend {
+		return nil
+	}
+
 	conn := b.open()
 	defer conn.Close()
 
@@ -207,6 +226,10 @@ func (b *Backend) SetStateSuccess(signature *tasks.Signature, results []*tasks.T
 
 // SetStateFailure updates task state to FAILURE
 func (b *Backend) SetStateFailure(signature *tasks.Signature, err string) error {
+	if signature.NoBackend {
+		return nil
+	}
+
 	conn := b.open()
 	defer conn.Close()
 

v1/brokers/azure/storage_queue.go

@@ -224,6 +224,12 @@ func (b *Broker) consumeOne(delivery azqueue.DequeueMessagesResponse, taskProces
 	sig.ReceivedAt = time.Now()
 	sig.AzureMessageContent = *msg.MessageText
 
+	if sig.UUID == "" {
+		// No UUID means the "task" was injected outside of machinery and we can't track state since there's no ID
+		// to track against.
+		sig.NoBackend = true
+	}
+
 	if msg.PopReceipt != nil {
 		sig.AzurePopReceipt = *msg.PopReceipt
 	}

v1/brokers/sqs/sqs.go

@@ -261,6 +261,12 @@ func (b *Broker) consumeOne(delivery *awssqs.ReceiveMessageOutput, taskProcessor
 
 	sig.ReceivedAt = time.Now()
 
+	if sig.UUID == "" {
+		// No UUID means the "task" was injected outside of machinery and we can't track state since there's no ID
+		// to track against.
+		sig.NoBackend = true
+	}
+
 	if delivery.Messages[0].ReceiptHandle != nil {
 		sig.SQSReceiptHandle = *delivery.Messages[0].ReceiptHandle
 	}

v1/server.go

@@ -229,6 +229,10 @@ func (server *Server) SendTaskWithContext(ctx context.Context, signature *tasks.
 		return nil, fmt.Errorf("Publish message error: %s", err)
 	}
 
+	if signature.NoBackend {
+		return nil, nil
+	}
+
 	return result.NewAsyncResult(signature, server.backend), nil
 }
 

v1/tasks/signature.go

@@ -74,6 +74,9 @@ type Signature struct {
 	AzurePopReceipt     string
 	AzureMessageContent string
 
+	// NoBackend: skip all operations related to machineries backend concept. Just leverage the broker abstraction.
+	NoBackend bool
+
 	// StopTaskDeletionOnError used with sqs when we want to send failed messages to dlq,
 	// and don't want machinery to delete from source queue
 	StopTaskDeletionOnError bool