Repository: beatlabs/patron
Pull request #110 (2019-11-24 15:57:12)

Async comp builder
## Which problem is this PR solving?

Async Component Builder Pattern #54

## Short description of the changes

- introduced aync component builder
(following the suggestion in the issue, backwards compatibility has been broken)
p.s. If this is not acceptable, we can keep the 'New' method as the component constructor, and use another name for the builder constructor
- introduced iota for enum of Failure Strategies and a map for values ,
to make comparisons regarding this enum easier
- extended test scope to cover ALL failure scenarios / error code paths
- refactor tests for async component, to make them more compact and readable
- re-used string templates for repeating log or error messages
- adjusted the existing examples to use the builder



Patch:
diff --git a/async/async.go b/async/async.go
index e06a096e0..942e6159f 100644
--- a/async/async.go
+++ b/async/async.go
@@ -9,6 +9,18 @@ import (
 	"github.com/beatlabs/patron/errors"
 )
 
+// FailStrategy type definition.
+type FailStrategy int
+
+const (
+	// NackExitStrategy does not acknowledge the message and exits the application on error.
+	NackExitStrategy FailStrategy = iota
+	// NackStrategy does not acknowledge the message, leaving it for reprocessing, and continues.
+	NackStrategy
+	// AckStrategy acknowledges message and continues.
+	AckStrategy
+)
+
 // ProcessorFunc definition of a async processor.
 type ProcessorFunc func(Message) error
 
diff --git a/async/component.go b/async/component.go
index 2e6610506..35d5554fd 100644
--- a/async/component.go
+++ b/async/component.go
@@ -2,6 +2,7 @@ package async
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/beatlabs/patron/errors"
@@ -9,6 +10,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+const propSetMSG = "property '%s' set for '%s'"
+
 var consumerErrors *prometheus.CounterVec
 
 func init() {
@@ -38,36 +41,86 @@ type Component struct {
 	retryWait    time.Duration
 }
 
-// New returns a new async component. The default behavior is to return a error of failure.
-// Use options to change the default behavior.
-func New(name string, p ProcessorFunc, cf ConsumerFactory, oo ...OptionFunc) (*Component, error) {
+// Builder gathers all required properties in order to construct a component
+type Builder struct {
+	errors       []error
+	name         string
+	proc         ProcessorFunc
+	failStrategy FailStrategy
+	cf           ConsumerFactory
+	retries      uint
+	retryWait    time.Duration
+}
 
+// New initializes a new builder for a component with the given name
+// by default the failStrategy will be NackExitStrategy.
+func New(name string, cf ConsumerFactory, proc ProcessorFunc) *Builder {
+	var errs []error
 	if name == "" {
-		return nil, errors.New("name is required")
+		errs = append(errs, errors.New("name is required"))
+	}
+	if cf == nil {
+		errs = append(errs, errors.New("consumer is required"))
 	}
+	if proc == nil {
+		errs = append(errs, errors.New("work processor is required"))
+	}
+	return &Builder{
+		name:   name,
+		cf:     cf,
+		proc:   proc,
+		errors: errs,
+	}
+}
 
-	if p == nil {
-		return nil, errors.New("work processor is required")
+// WithFailureStrategy defines the failure strategy to be used
+// default value is NackExitStrategy
+// it will append an error to the builder if the strategy is not one of the pre-defined ones.
+func (cb *Builder) WithFailureStrategy(fs FailStrategy) *Builder {
+	if fs > AckStrategy || fs < NackExitStrategy {
+		cb.errors = append(cb.errors, errors.New("invalid strategy provided"))
+	} else {
+		log.Infof(propSetMSG, "failure strategy", cb.name)
+		cb.failStrategy = fs
 	}
+	return cb
+}
 
-	if cf == nil {
-		return nil, errors.New("consumer is required")
+// WithRetries specifies the retry events number for the component
+// default value is '0'.
+func (cb *Builder) WithRetries(retries uint) *Builder {
+	log.Infof(propSetMSG, "retries", cb.name)
+	cb.retries = retries
+	return cb
+}
+
+// WithRetryWait specifies the duration for the component to wait between retries
+// default value is '0'
+// it will append an error to the builder if the value is smaller than '0'.
+func (cb *Builder) WithRetryWait(retryWait time.Duration) *Builder {
+	if retryWait < 0 {
+		cb.errors = append(cb.errors, errors.New("invalid retry wait provided"))
+	} else {
+		log.Infof(propSetMSG, "retryWait", cb.name)
+		cb.retryWait = retryWait
 	}
+	return cb
+}
 
-	c := &Component{
-		name:         name,
-		proc:         p,
-		cf:           cf,
-		failStrategy: NackExitStrategy,
-		retries:      0,
-		retryWait:    0,
+// Create constructs the Component applying
+func (cb *Builder) Create() (*Component, error) {
+
+	if len(cb.errors) > 0 {
+		return nil, errors.Aggregate(cb.errors...)
 	}
 
-	for _, o := range oo {
-		err := o(c)
-		if err != nil {
-			return nil, err
-		}
+	c := &Component{
+		name:         cb.name,
+		proc:         cb.proc,
+		cf:           cb.cf,
+		failStrategy: cb.failStrategy,
+		retries:      int(cb.retries),
+		retryWait:    cb.retryWait,
 	}
 
 	return c, nil
@@ -148,7 +201,10 @@ func (c *Component) processMessage(msg Message, ch chan error) {
 	}
 }
 
+var errInvalidFS = errors.New("invalid failure strategy")
+
 func (c *Component) executeFailureStrategy(msg Message, err error) error {
+	const failureStrategyErrorMSG = "%s failed when executing failure strategy"
 	log.FromContext(msg.Context()).Errorf("failed to process message, failure strategy executed: %v", err)
 	switch c.failStrategy {
 	case NackExitStrategy:
@@ -156,15 +212,15 @@ func (c *Component) executeFailureStrategy(msg Message, err error) error {
 	case NackStrategy:
 		err := msg.Nack()
 		if err != nil {
-			return errors.Wrap(err, "nack failed when executing failure strategy")
+			return errors.Wrap(err, fmt.Sprintf(failureStrategyErrorMSG, "nack"))
 		}
 	case AckStrategy:
 		err := msg.Ack()
 		if err != nil {
-			return errors.Wrap(err, "ack failed when executing failure strategy")
+			return errors.Wrap(err, fmt.Sprintf(failureStrategyErrorMSG, "ack"))
 		}
 	default:
-		return errors.New("invalid failure strategy")
+		return errInvalidFS
 	}
 	return nil
 }
diff --git a/async/option.go b/async/option.go
deleted file mode 100644
index f7a61266e..000000000
--- a/async/option.go
+++ /dev/null
@@ -1,66 +0,0 @@
-package async
-
-import (
-	"time"
-
-	"github.com/beatlabs/patron/errors"
-	"github.com/beatlabs/patron/log"
-)
-
-// FailStrategy type definition.
-type FailStrategy int
-
-const (
-	// NackExitStrategy does not acknowledge the message and exits the application on error.
-	NackExitStrategy FailStrategy = 0
-	// NackStrategy does not acknowledge the message, leaving it for reprocessing, and continues.
-	NackStrategy FailStrategy = 1
-	// AckStrategy acknowledges message and continues.
-	AckStrategy FailStrategy = 2
-)
-
-func (fs FailStrategy) String() string {
-	switch fs {
-	case NackExitStrategy:
-		return "NackExitStrategy"
-	case NackStrategy:
-		return "NackStrategy"
-	case AckStrategy:
-		return "AckStrategy"
-	default:
-		return "N/A"
-	}
-}
-
-// OptionFunc definition for configuring the component in a functional way.
-type OptionFunc func(*Component) error
-
-// FailureStrategy option for setting the strategy of handling failures in the async component.
-func FailureStrategy(fs FailStrategy) OptionFunc {
-	return func(c *Component) error {
-		if fs > AckStrategy || fs < NackExitStrategy {
-			return errors.New("invalid strategy provided")
-		}
-		c.failStrategy = fs
-		log.Info("failure strategy set")
-		return nil
-	}
-}
-
-// ConsumerRetry set's the parameters for the retry policy of the consumer.
-func ConsumerRetry(retries int, retryWait time.Duration) OptionFunc {
-	return func(c *Component) error {
-		if retries < 0 {
-			return errors.New("invalid retries provided")
-		}
-
-		if retryWait < 0 {
-			return errors.New("invalid retry wait provided")
-		}
-
-		c.retries = retries
-		c.retryWait = retryWait
-		log.Info("consumer retry set")
-		return nil
-	}
-}
diff --git a/examples/fifth/main.go b/examples/fifth/main.go
index 8d2bb2fd8..de59b08f3 100644
--- a/examples/fifth/main.go
+++ b/examples/fifth/main.go
@@ -97,7 +97,10 @@ func createSQSComponent(api sqsiface.SQSAPI) (*sqsComponent, error) {
 		return nil, err
 	}
 
-	cmp, err := async.New("sqs-cmp", sqsCmp.Process, cf, async.ConsumerRetry(10, 10*time.Second))
+	cmp, err := async.New("sqs-cmp", cf, sqsCmp.Process).
+		WithRetries(10).
+		WithRetryWait(10 * time.Second).
+		Create()
 	if err != nil {
 		return nil, err
 	}
diff --git a/examples/fourth/main.go b/examples/fourth/main.go
index 9f8e66cdb..a51bb6edf 100644
--- a/examples/fourth/main.go
+++ b/examples/fourth/main.go
@@ -201,7 +201,10 @@ func newAmqpComponent(url, queue, exchangeName, exchangeType string, bindings []
 		return nil, err
 	}
 
-	cmp, err := async.New("amqp-cmp", amqpCmp.Process, cf, async.ConsumerRetry(10, 10*time.Second))
+	cmp, err := async.New("amqp-cmp", cf, amqpCmp.Process).
+		WithRetries(10).
+		WithRetryWait(10 * time.Second).
+		Create()
 	if err != nil {
 		return nil, err
 	}
diff --git a/examples/third/main.go b/examples/third/main.go
index a7ca38957..85193b1bb 100644
--- a/examples/third/main.go
+++ b/examples/third/main.go
@@ -86,7 +86,10 @@ func newKafkaComponent(name, broker, topic, group, amqpURL, amqpExc string) (*ka
 		return nil, err
 	}
 
-	cmp, err := async.New("kafka-cmp", kafkaCmp.Process, cf, async.ConsumerRetry(10, 5*time.Second))
+	cmp, err := async.New("kafka-cmp", cf, kafkaCmp.Process).
+		WithRetries(10).
+		WithRetryWait(5 * time.Second).
+		Create()
 	if err != nil {
 		return nil, err
 	}


Test patch:
diff --git a/async/component_test.go b/async/component_test.go
index e1a85211b..5eac937fc 100644
--- a/async/component_test.go
+++ b/async/component_test.go
@@ -2,20 +2,23 @@ package async
 
 import (
 	"context"
+	"errors"
+	"strings"
 	"testing"
 	"time"
 
-	"github.com/beatlabs/patron/errors"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestNew(t *testing.T) {
 	proc := mockProcessor{}
 	type args struct {
-		name string
-		p    ProcessorFunc
-		cf   ConsumerFactory
-		opt  OptionFunc
+		name      string
+		p         ProcessorFunc
+		cf        ConsumerFactory
+		fs        FailStrategy
+		retries   uint
+		retryWait time.Duration
 	}
 	tests := []struct {
 		name    string
@@ -24,33 +27,42 @@ func TestNew(t *testing.T) {
 	}{
 		{
 			name:    "success",
-			args:    args{name: "name", p: proc.Process, cf: &mockConsumerFactory{}, opt: FailureStrategy(NackExitStrategy)},
+			args:    args{name: "name", p: proc.Process, cf: &mockConsumerFactory{}, fs: NackExitStrategy},
 			wantErr: false,
 		},
 		{
 			name:    "failed, missing name",
-			args:    args{name: "", p: proc.Process, cf: &mockConsumerFactory{}, opt: FailureStrategy(NackExitStrategy)},
+			args:    args{name: "", p: proc.Process, cf: &mockConsumerFactory{}, fs: NackExitStrategy},
 			wantErr: true,
 		},
 		{
 			name:    "failed, missing processor func",
-			args:    args{name: "name", p: nil, cf: &mockConsumerFactory{}, opt: FailureStrategy(NackExitStrategy)},
+			args:    args{name: "name", p: nil, cf: &mockConsumerFactory{}, fs: NackExitStrategy},
 			wantErr: true,
 		},
 		{
 			name:    "failed, missing consumer",
-			args:    args{name: "name", p: proc.Process, cf: nil, opt: FailureStrategy(NackExitStrategy)},
+			args:    args{name: "name", p: proc.Process, cf: nil, fs: NackExitStrategy},
 			wantErr: true,
 		},
 		{
 			name:    "failed, invalid fail strategy",
-			args:    args{name: "name", p: proc.Process, cf: &mockConsumerFactory{}, opt: FailureStrategy(3)},
+			args:    args{name: "name", p: proc.Process, cf: &mockConsumerFactory{}, fs: 3},
+			wantErr: true,
+		},
+		{
+			name:    "failed, invalid retry retry timeout",
+			args:    args{name: "name", p: proc.Process, cf: &mockConsumerFactory{}, retryWait: -2},
 			wantErr: true,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := New(tt.args.name, tt.args.p, tt.args.cf, tt.args.opt)
+			got, err := New(tt.args.name, tt.args.cf, tt.args.p).
+				WithFailureStrategy(tt.args.fs).
+				WithRetries(tt.args.retries).
+				WithRetryWait(tt.args.retryWait).
+				Create()
 			if tt.wantErr {
 				assert.Error(t, err)
 				assert.Nil(t, got)
@@ -62,174 +74,412 @@ func TestNew(t *testing.T) {
 	}
 }
 
-func TestRun_ReturnsError(t *testing.T) {
-	cnr := mockConsumer{consumeError: true}
-	proc := mockProcessor{}
-	cmp, err := New("test", proc.Process, &mockConsumerFactory{c: &cnr})
+type proxyBuilder struct {
+	proc      mockProcessor
+	cnr       mockConsumer
+	cf        ConsumerFactory
+	fs        FailStrategy
+	retries   int
+	retryWait time.Duration
+}
+
+func run(ctx context.Context, t *testing.T, builder *proxyBuilder) error {
+
+	if builder.cf == nil {
+		builder.cf = &mockConsumerFactory{c: &builder.cnr}
+	}
+
+	cmp, err := New("test", builder.cf, builder.proc.Process).
+		WithFailureStrategy(builder.fs).
+		WithRetries(uint(builder.retries)).
+		WithRetryWait(builder.retryWait).
+		Create()
 	assert.NoError(t, err)
-	ctx := context.Background()
-	err = cmp.Run(ctx)
+	return cmp.Run(ctx)
+}
+
+// TestRun_ReturnsError expects a consumer consume Error
+func TestRun_ReturnsError(t *testing.T) {
+
+	builder := proxyBuilder{
+		cnr: mockConsumer{consumeError: true},
+	}
+	err := run(context.Background(), t, &builder)
+
 	assert.Error(t, err)
+	assert.True(t, strings.Contains(err.Error(), errConsumer.Error()))
+	assert.Equal(t, 0, builder.proc.execs)
+
 }
 
+// TestRun_WithCancel_CloseError expects a consumer closing error
+func TestRun_WithCancel_CloseError(t *testing.T) {
+
+	builder := proxyBuilder{
+		cnr: mockConsumer{clsError: true},
+	}
+
+	ctx, cnl := context.WithCancel(context.Background())
+	cnl()
+
+	err := run(ctx, t, &builder)
+
+	assert.Error(t, err)
+	assert.Equal(t, errConsumerClose, err)
+	assert.Equal(t, 0, builder.proc.execs)
+
+}
+
+// TestRun_Process_Error_NackExitStrategy expects a PROC ERROR
+// from an error producing processor
+// which will cause the component to return with an error
+// due to the Nack FailureStrategy
 func TestRun_Process_Error_NackExitStrategy(t *testing.T) {
-	cnr := mockConsumer{
-		chMsg: make(chan Message, 10),
-		chErr: make(chan error, 10),
+
+	builder := proxyBuilder{
+		proc: mockProcessor{errReturn: true},
+		cnr: mockConsumer{
+			chMsg: make(chan Message, 10),
+			chErr: make(chan error, 10),
+		},
 	}
-	proc := mockProcessor{retError: true}
-	cmp, err := New("test", proc.Process, &mockConsumerFactory{c: &cnr})
-	assert.NoError(t, err)
+
 	ctx := context.Background()
-	cnr.chMsg <- &mockMessage{ctx: ctx}
-	err = cmp.Run(ctx)
+	builder.cnr.chMsg <- &mockMessage{ctx: ctx}
+
+	err := run(ctx, t, &builder)
+
 	assert.Error(t, err)
+	assert.True(t, strings.Contains(err.Error(), errProcess.Error()))
+	assert.Equal(t, 1, builder.proc.execs)
+
 }
 
+// TestRun_Process_Error_NackStrategy expects a PROC ERROR
+// from an error producing processor
+// but due to the Nack FailureStrategy, it will continue processing other messages
 func TestRun_Process_Error_NackStrategy(t *testing.T) {
-	cnr := mockConsumer{
-		chMsg: make(chan Message, 10),
-		chErr: make(chan error, 10),
+
+	builder := proxyBuilder{
+		proc: mockProcessor{errReturn: true},
+		cnr: mockConsumer{
+			chMsg: make(chan Message, 10),
+			chErr: make(chan error, 10),
+		},
+		fs: NackStrategy,
 	}
-	proc := mockProcessor{retError: true}
-	cmp, err := New("test", proc.Process, &mockConsumerFactory{c: &cnr}, FailureStrategy(NackStrategy))
-	assert.NoError(t, err)
+
 	ctx, cnl := context.WithCancel(context.Background())
-	cnr.chMsg <- &mockMessage{ctx: ctx}
+	builder.cnr.chMsg <- &mockMessage{ctx: ctx}
 	ch := make(chan bool)
 	go func() {
-		assert.NoError(t, cmp.Run(ctx))
+		assert.NoError(t, run(ctx, t, &builder))
 		ch <- true
 	}()
 	time.Sleep(10 * time.Millisecond)
 	cnl()
-	assert.True(t, <-ch)
+
+	select {
+	case _, ok := <-builder.cnr.chErr:
+		if ok {
+			assert.Fail(t, "we dont expect an error , given our nack failure strategy setup")
+		}
+	default:
+		assert.True(t, <-ch)
+	}
+	assert.Equal(t, 1, builder.proc.execs)
+
+}
+
+// TestRun_ProcessError_WithNackError expects a PROC ERROR
+// from an error producing processor
+// but also a Nack Error from the message
+// This will cause the component to stop execution, as it cannot execute the Nack failure strategy
+func TestRun_ProcessError_WithNackError(t *testing.T) {
+
+	builder := proxyBuilder{
+		proc: mockProcessor{errReturn: true},
+		cnr: mockConsumer{
+			chMsg: make(chan Message, 10),
+			chErr: make(chan error, 10),
+		},
+		fs: NackStrategy,
+	}
+
+	ctx := context.Background()
+	builder.cnr.chMsg <- &mockMessage{ctx: ctx, nackError: true}
+
+	err := run(ctx, t, &builder)
+
+	assert.Error(t, err)
+	assert.True(t, strings.Contains(err.Error(), errNack.Error()))
+	assert.Equal(t, 1, builder.proc.execs)
+
 }
 
+// TestRun_Process_Error_AckStrategy expects a PROC ERROR
+// from an error producing processor
+// but due to the Ack FailureStrategy, it will continue processing other messages
 func TestRun_Process_Error_AckStrategy(t *testing.T) {
-	cnr := mockConsumer{
-		chMsg: make(chan Message, 10),
-		chErr: make(chan error, 10),
+
+	builder := proxyBuilder{
+		proc: mockProcessor{errReturn: true},
+		cnr: mockConsumer{
+			chMsg: make(chan Message, 10),
+			chErr: make(chan error, 10),
+		},
+		fs: NackStrategy,
 	}
-	proc := mockProcessor{retError: true}
-	cmp, err := New("test", proc.Process, &mockConsumerFactory{c: &cnr}, FailureStrategy(AckStrategy))
-	assert.NoError(t, err)
+
 	ctx, cnl := context.WithCancel(context.Background())
-	cnr.chMsg <- &mockMessage{ctx: ctx}
+	builder.cnr.chMsg <- &mockMessage{ctx: ctx}
 	ch := make(chan bool)
 	go func() {
-		assert.NoError(t, cmp.Run(ctx))
+		assert.NoError(t, run(ctx, t, &builder))
 		ch <- true
 	}()
 	time.Sleep(10 * time.Millisecond)
 	cnl()
-	assert.True(t, <-ch)
+
+	select {
+	case _, ok := <-builder.cnr.chErr:
+		if ok {
+			assert.Fail(t, "we dont expect an error , given our ack failure strategy setup")
+		}
+	default:
+		assert.True(t, <-ch)
+	}
+	assert.Equal(t, 1, builder.proc.execs)
+
 }
 
-func TestRun_Process_Error_InvalidStrategy(t *testing.T) {
-	cnr := mockConsumer{
-		chMsg: make(chan Message, 10),
-		chErr: make(chan error, 10),
+// TestRun_ProcessError_WithAckError expects a PROC ERROR
+// from an error producing processor
+// but also an Ack Error from the message
+// This will cause the component to stop execution, as it cannot execute the Nack failure strategy
+func TestRun_ProcessError_WithAckError(t *testing.T) {
+
+	builder := proxyBuilder{
+		proc: mockProcessor{errReturn: true},
+		cnr: mockConsumer{
+			chMsg: make(chan Message, 10),
+			chErr: make(chan error, 10),
+		},
+		fs: AckStrategy,
 	}
-	proc := mockProcessor{retError: true}
-	cmp, err := New("test", proc.Process, &mockConsumerFactory{c: &cnr})
-	cmp.failStrategy = 4
-	assert.NoError(t, err)
+
 	ctx := context.Background()
-	cnr.chMsg <- &mockMessage{ctx: ctx}
-	err = cmp.Run(ctx)
+	builder.cnr.chMsg <- &mockMessage{ctx: ctx, ackError: true}
+
+	err := run(ctx, t, &builder)
+
+	assert.Error(t, err)
+	assert.True(t, strings.Contains(err.Error(), errAck.Error()))
+	assert.Equal(t, 1, builder.proc.execs)
+
+}
+
+// TestRun_MessageAckError expects an ack error from the message acknowledgement
+// it will break the execution of the component due to the default NackExit failure strategy
+func TestRun_MessageAckError(t *testing.T) {
+
+	builder := proxyBuilder{
+		cnr: mockConsumer{
+			chMsg: make(chan Message, 10),
+			chErr: make(chan error, 10),
+		},
+	}
+
+	ctx := context.Background()
+	builder.cnr.chMsg <- &mockMessage{ctx: ctx, ackError: true}
+	err := run(ctx, t, &builder)
+
 	assert.Error(t, err)
+	assert.Equal(t, errAck, err)
+	assert.Equal(t, 1, builder.proc.execs)
+
 }
 
+// TestRun_ConsumeError will break the component execution,
+// when an error is injected into the consumers error channel
+// while using the default NackExit Failure Strategy
 func TestRun_ConsumeError(t *testing.T) {
-	cnr := mockConsumer{
-		chMsg: make(chan Message, 10),
-		chErr: make(chan error, 10),
+
+	builder := proxyBuilder{
+		cnr: mockConsumer{
+			chMsg: make(chan Message, 10),
+			chErr: make(chan error, 10),
+		},
 	}
-	proc := mockProcessor{retError: true}
-	cmp, err := New("test", proc.Process, &mockConsumerFactory{c: &cnr})
-	assert.NoError(t, err)
+
 	ctx := context.Background()
-	cnr.chErr <- errors.New("CONSUMER ERROR")
-	err = cmp.Run(ctx)
+	builder.cnr.chErr <- errConsumer
+	err := run(ctx, t, &builder)
+
 	assert.Error(t, err)
+	assert.True(t, strings.Contains(err.Error(), errConsumer.Error()))
+	assert.Equal(t, 0, builder.proc.execs)
+
 }
 
+// TestRun_ConsumeError_WithRetry will retry the specified amount of times
+// before exiting the execution
 func TestRun_ConsumeError_WithRetry(t *testing.T) {
-	proc := mockProcessor{retError: true}
-	cmp, err := New("test", proc.Process, &mockConsumerFactory{retErr: true}, ConsumerRetry(3, 2*time.Millisecond))
-	assert.NoError(t, err)
-	ctx := context.Background()
-	err = cmp.Run(ctx)
+
+	retries := 3
+	cf := &mockConsumerFactory{errRet: true}
+	builder := proxyBuilder{
+		cf:        cf,
+		retries:   retries,
+		retryWait: 2 * time.Millisecond,
+	}
+
+	err := run(context.Background(), t, &builder)
+
+	assert.Error(t, err)
+	assert.True(t, retries <= cf.execs)
+	assert.Equal(t, 0, builder.proc.execs)
+
+}
+
+// TestRun_ConsumeError_WithRetry_AndContextCancel will retry after a consumer error
+// only a small amount fo times, due to the context being canceled as well
+func TestRun_ConsumeError_WithRetry_AndContextCancel(t *testing.T) {
+
+	retries := 33
+	cf := &mockConsumerFactory{errRet: true}
+	builder := proxyBuilder{
+		cf:        cf,
+		retries:   retries,
+		retryWait: 2 * time.Millisecond,
+	}
+
+	ctx, cnl := context.WithCancel(context.Background())
+	cnl()
+	err := run(ctx, t, &builder)
+
 	assert.Error(t, err)
+	assert.Equal(t, ctx.Err(), context.Canceled)
+	assert.True(t, retries > cf.execs)
+	assert.Equal(t, 0, builder.proc.execs)
+
 }
 
+// TestRun_Process_Shutdown verifies the process shuts down on a context cancellation action
 func TestRun_Process_Shutdown(t *testing.T) {
-	cnr := mockConsumer{
-		chMsg: make(chan Message, 10),
-		chErr: make(chan error, 10),
+
+	builder := proxyBuilder{
+		cnr: mockConsumer{
+			chMsg: make(chan Message, 10),
+			chErr: make(chan error, 10),
+		},
 	}
-	proc := mockProcessor{retError: false}
-	cmp, err := New("test", proc.Process, &mockConsumerFactory{c: &cnr})
-	assert.NoError(t, err)
-	cnr.chMsg <- &mockMessage{ctx: context.Background()}
+
+	builder.cnr.chMsg <- &mockMessage{ctx: context.Background()}
 	ch := make(chan bool)
 	ctx, cnl := context.WithCancel(context.Background())
 	go func() {
-		err1 := cmp.Run(ctx)
+		err1 := run(ctx, t, &builder)
 		assert.NoError(t, err1)
 		ch <- true
 	}()
 	time.Sleep(10 * time.Millisecond)
 	cnl()
+
 	assert.True(t, <-ch)
+	assert.Equal(t, 1, builder.proc.execs)
+
+}
+
+// TestRun_Process_Error_InvalidStrategy expects a invalid failure strategy error
+// NOTE : we injected the failure strategy after the construction,
+// in order to avoid the failure strategy check
+func TestRun_Process_Error_InvalidStrategy(t *testing.T) {
+	cnr := mockConsumer{
+		chMsg: make(chan Message, 10),
+		chErr: make(chan error, 10),
+	}
+	proc := mockProcessor{errReturn: true}
+	cmp, err := New("test", &mockConsumerFactory{c: &cnr}, proc.Process).
+		Create()
+	assert.NoError(t, err)
+	cmp.failStrategy = 4
+	ctx := context.Background()
+	cnr.chMsg <- &mockMessage{ctx: ctx}
+	err = cmp.Run(ctx)
+	assert.Error(t, err)
+	assert.Equal(t, errInvalidFS, err)
+	assert.Equal(t, 1, proc.execs)
+
 }
 
 type mockMessage struct {
-	ctx context.Context
+	ctx       context.Context
+	ackError  bool
+	nackError bool
 }
 
 func (mm *mockMessage) Context() context.Context {
 	return mm.ctx
 }
 
+// Decode is not called in our tests, because the mockProcessor will ignore the message decoding
 func (mm *mockMessage) Decode(v interface{}) error {
 	return nil
 }
 
+var errAck = errors.New("MESSAGE ACK ERROR")
+
 func (mm *mockMessage) Ack() error {
+	if mm.ackError {
+		return errAck
+	}
 	return nil
 }
 
+var errNack = errors.New("MESSAGE NACK ERROR")
+
 func (mm *mockMessage) Nack() error {
+	if mm.nackError {
+		return errNack
+	}
 	return nil
 }
 
 type mockProcessor struct {
-	retError bool
+	errReturn bool
+	execs     int
 }
 
+var errProcess = errors.New("PROC ERROR")
+
 func (mp *mockProcessor) Process(msg Message) error {
-	if mp.retError {
-		return errors.New("PROC ERROR")
+	mp.execs++
+	if mp.errReturn {
+		return errProcess
 	}
 	return nil
 }
 
 type mockConsumerFactory struct {
 	c      Consumer
-	retErr bool
+	errRet bool
+	execs  int
 }
 
+var errFactory = errors.New("FACTORY ERROR")
+
 func (mcf *mockConsumerFactory) Create() (Consumer, error) {
-	if mcf.retErr {
-		return nil, errors.New("FACTORY ERROR")
+	mcf.execs++
+	if mcf.errRet {
+		return nil, errFactory
 	}
 	return mcf.c, nil
 }
 
 type mockConsumer struct {
 	consumeError bool
+	clsError     bool
 	chMsg        chan Message
 	chErr        chan error
 }
@@ -237,13 +487,20 @@ type mockConsumer struct {
 func (mc *mockConsumer) SetTimeout(timeout time.Duration) {
 }
 
+var errConsumer = errors.New("CONSUMER ERROR")
+
 func (mc *mockConsumer) Consume(context.Context) (<-chan Message, <-chan error, error) {
 	if mc.consumeError {
-		return nil, nil, errors.New("CONSUMER ERROR")
+		return nil, nil, errConsumer
 	}
 	return mc.chMsg, mc.chErr, nil
 }
 
+var errConsumerClose = errors.New("CONSUMER CLOSE ERROR")
+
 func (mc *mockConsumer) Close() error {
+	if mc.clsError {
+		return errConsumerClose
+	}
 	return nil
 }
diff --git a/async/option_test.go b/async/option_test.go
deleted file mode 100644
index 50a3d468e..000000000
--- a/async/option_test.go
+++ /dev/null
@@ -1,83 +0,0 @@
-package async
-
-import (
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestFailStrategy_String(t *testing.T) {
-	tests := []struct {
-		name string
-		fs   FailStrategy
-		want string
-	}{
-		{name: "NackExitStrategy", fs: NackExitStrategy, want: "NackExitStrategy"},
-		{name: "NackStrategy", fs: NackStrategy, want: "NackStrategy"},
-		{name: "AckStrategy", fs: AckStrategy, want: "AckStrategy"},
-		{name: "Not mapped", fs: -1, want: "N/A"},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			assert.Equal(t, tt.want, tt.fs.String())
-		})
-	}
-}
-
-func TestFailureStrategy(t *testing.T) {
-	proc := mockProcessor{}
-	type args struct {
-		fs FailStrategy
-	}
-	tests := []struct {
-		name    string
-		args    args
-		wantErr bool
-	}{
-		{name: "success", args: args{fs: NackExitStrategy}, wantErr: false},
-		{name: "invalid strategy (lower)", args: args{fs: -1}, wantErr: true},
-		{name: "invalid strategy (higher)", args: args{fs: 3}, wantErr: true},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			c, err := New("test", proc.Process, &mockConsumerFactory{})
-			assert.NoError(t, err)
-			err = FailureStrategy(tt.args.fs)(c)
-			if tt.wantErr {
-				assert.Error(t, err)
-			} else {
-				assert.NoError(t, err)
-			}
-		})
-	}
-}
-
-func TestConsumerRetry(t *testing.T) {
-	proc := mockProcessor{}
-	type args struct {
-		retries   int
-		retryWait time.Duration
-	}
-	tests := []struct {
-		name    string
-		args    args
-		wantErr bool
-	}{
-		{name: "success", args: args{retries: 3, retryWait: time.Second}, wantErr: false},
-		{name: "invalid retries", args: args{retries: -1, retryWait: time.Second}, wantErr: true},
-		{name: "invalid retry wait", args: args{retries: 3, retryWait: -1}, wantErr: true},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			c, err := New("test", proc.Process, &mockConsumerFactory{})
-			assert.NoError(t, err)
-			err = ConsumerRetry(tt.args.retries, tt.args.retryWait)(c)
-			if tt.wantErr {
-				assert.Error(t, err)
-			} else {
-				assert.NoError(t, err)
-			}
-		})
-	}
-}