feat: add batch response to support partial failures

BREAKING CHANGE: multiMessageHandler renamed to batchHandler. It now requires a BatchResponse

Author: francescopepe

README.md

@@ -6,7 +6,7 @@ Formigo is a powerful and flexible Golang library designed to simplify the proce
 
 - **Efficient Throughput Management**: it offers optimal throughput management, allowing you to fine-tune the number of Go routines responsible for both polling messages from the queue and processing them. This dynamic control ensures maximum efficiency in various scenarios, making the library highly adaptable to your application's needs.
 
-- **Configurable Batch Processing**: it uses powerful batch processing capabilities, enabling you to handle messages efficiently in customizable batches. With the Multiple Message Handler, messages can be processed in batches of a size you define, granting you full control over the processing logic. Moreover, you can adjust the batch buffer size and timeout settings, providing a flexible and optimal solution to process messages under various workloads.
+- **Configurable Batch Processing**: it uses powerful batch processing capabilities, enabling you to handle messages efficiently in customizable batches. With the Batch Handler, messages can be processed in batches of a size you define, granting you full control over the processing logic. Moreover, you can adjust the batch buffer size and timeout settings, providing a flexible and optimal solution to process messages under various workloads.
 
 - **Context Cancellation**: Effortlessly stop the QueueWorker by canceling its context. This feature guarantees smooth and controlled termination of the worker whenever required.
 
@@ -140,8 +140,8 @@ func main() {
     wkr := formigo.NewWorker(formigo.Configuration{
         Client: sqsClient,
         Concurrency: 100,
-        Consumer: formigo.NewMultiMessageConsumer(formigo.MultiMessageConsumerConfiguration{
-            BufferConfig: formigo.MultiMessageBufferConfiguration{
+        Consumer: formigo.BatchConsumer(formigo.BatchConsumerConfiguration{
+            BufferConfig: formigo.BatchBufferConfiguration{
                 Size:    100,
                 Timeout: time.Second * 5,
             },
@@ -180,7 +180,7 @@ By processing messages in batches, the worker can significantly enhance throughp
 | Concurrency   | Number of Go routines that process the messages from the Queue. Higher values are useful for slow I/O operations in the consumer's handler.     | 100           |
 | Retrievers    | Number of Go routines that retrieve messages from the Queue. Higher values are helpful for slow networks or when consumers are quicker.         | 1             |
 | ErrorConfig   | Defines the error threshold and interval for worker termination and error reporting function.                                                   | None          |
-| Consumer      | The message consumer, either MessageConsumer or MultipleMessageConsumer.                                                                        | None          |
+| Consumer      | The message consumer, either MessageConsumer or BatchConsumer.                                                                                  | None          |
 
 ## License
 

config.go

@@ -37,9 +37,9 @@ type ErrorConfiguration struct {
 	ReportFunc func(err error) bool
 }
 
-// The MultiMessageBufferConfiguration defines a buffer which is consumed by the worker when either
+// The BatchConsumerBufferConfiguration defines a buffer which is consumed by the worker when either
 // the buffer is full or the timeout has passed since the first message got added.
-type MultiMessageBufferConfiguration struct {
+type BatchConsumerBufferConfiguration struct {
 	// Max number of messages that the buffer can contain.
 	// Default: 10.
 	Size int
@@ -56,9 +56,9 @@ type MessageConsumerConfiguration struct {
 	Handler messageHandler
 }
 
-type MultiMessageConsumerConfiguration struct {
-	Handler      multiMessageHandler
-	BufferConfig MultiMessageBufferConfiguration
+type BatchConsumerConfiguration struct {
+	Handler      batchHandler
+	BufferConfig BatchConsumerBufferConfiguration
 }
 
 type Configuration struct {

consumers.go

@@ -10,9 +10,12 @@ import (
 	"github.com/francescopepe/formigo/internal/messages"
 )
 
-type singleMessageHandler = func(ctx context.Context, msg Message) error
-type multiMessageHandler = func(ctx context.Context, msgs []Message) error
+type BatchResponse struct {
+	FailedMessagesId []interface{}
+}
+
 type messageHandler = func(ctx context.Context, msg Message) error
+type batchHandler = func(ctx context.Context, msgs []Message) (BatchResponse, error)
 
 // This means that the buffered messages didn't get passed to the handler within
 // the first message's timeout.
@@ -103,39 +106,43 @@ func NewMessageConsumer(config MessageConsumerConfiguration) *messageConsumer {
 	}
 }
 
-// multiMessageConsumer allows to process multiple messages at a time. This can be useful
+// batchConsumer allows to process multiple messages at a time. This can be useful
 // for batch updates or use cases with high throughput.
-type multiMessageConsumer struct {
-	handler      multiMessageHandler
-	bufferConfig MultiMessageBufferConfiguration
+type batchConsumer struct {
+	handler      batchHandler
+	bufferConfig BatchConsumerBufferConfiguration
 }
 
 // It processes the messages and push them downstream for deletion.
-func (c *multiMessageConsumer) processMessages(ctrl *controller, deleteCh chan<- messages.Message, ctx context.Context, msgs []messages.Message) {
-	err := wrapHandler(func() error {
-		// Convert slice to the abstraction
-		converted := make([]Message, len(msgs))
-		for _, msg := range msgs {
-			converted = append(converted, msg)
+func (c *batchConsumer) processMessages(ctrl *controller, deleteCh chan<- messages.Message, ctx context.Context, msgs []messages.Message) {
+	defer func() {
+		if r := recover(); r != nil {
+			ctrl.reportError(fmt.Errorf("panic error: %s", r))
 		}
+	}()
 
-		return c.handler(ctx, converted)
-	})
+	// Convert slice to the abstraction
+	converted := make([]Message, 0, len(msgs))
+	for _, msg := range msgs {
+		converted = append(converted, msg)
+	}
+
+	resp, err := c.handler(ctx, converted)
 	if err != nil {
-		ctrl.reportError(fmt.Errorf("failed to process messages: %w", err))
-		return
+		ctrl.reportError(fmt.Errorf("failed to process batch: %w", err))
 	}
 
+	toDelete := c.buildMessagesToDeleteFromBatchResponse(msgs, resp)
 	// Push messages for deletion
-	for _, msg := range msgs {
+	for _, msg := range toDelete {
 		deleteCh <- msg
 	}
 }
 
 // Consumes and deletes a number of messages in the interval [1, N] based on configuration
 // provided in the BufferConfiguration.
 // It stops only when the messageCh gets closed and doesn't have any messages in it.
-func (c *multiMessageConsumer) consume(concurrency int, ctrl *controller, messageCh <-chan messages.Message, deleteCh chan<- messages.Message) {
+func (c *batchConsumer) consume(concurrency int, ctrl *controller, messageCh <-chan messages.Message, deleteCh chan<- messages.Message) {
 	consumers := makeAvailableConsumers(concurrency)
 
 	// Create buffer
@@ -211,7 +218,28 @@ func (c *multiMessageConsumer) consume(concurrency int, ctrl *controller, messag
 	wg.Wait()
 }
 
-func NewMultiMessageConsumer(config MultiMessageConsumerConfiguration) *multiMessageConsumer {
+func (c *batchConsumer) buildMessagesToDeleteFromBatchResponse(msgs []messages.Message, resp BatchResponse) []messages.Message {
+	if len(resp.FailedMessagesId) == 0 {
+		return msgs
+	}
+
+	toDelete := make([]messages.Message, 0, len(msgs))
+
+	failedMessagesIdIndexed := make(map[interface{}]struct{}, len(resp.FailedMessagesId))
+	for _, id := range resp.FailedMessagesId {
+		failedMessagesIdIndexed[id] = struct{}{}
+	}
+
+	for _, msg := range msgs {
+		if _, ok := failedMessagesIdIndexed[msg.Id()]; !ok {
+			toDelete = append(toDelete, msg)
+		}
+	}
+
+	return toDelete
+}
+
+func NewBatchConsumer(config BatchConsumerConfiguration) *batchConsumer {
 	if config.BufferConfig.Size == 0 {
 		config.BufferConfig.Size = 10
 	}
@@ -220,15 +248,14 @@ func NewMultiMessageConsumer(config MultiMessageConsumerConfiguration) *multiMes
 		config.BufferConfig.Timeout = time.Second
 	}
 
-	return &multiMessageConsumer{
+	return &batchConsumer{
 		handler:      config.Handler,
 		bufferConfig: config.BufferConfig,
 	}
 }
 
 // Interface guards
 var (
-	_ consumer = (*singleMessageConsumer)(nil)
-	_ consumer = (*multiMessageConsumer)(nil)
 	_ consumer = (*messageConsumer)(nil)
+	_ consumer = (*batchConsumer)(nil)
 )

internal/messages/messages.go

@@ -9,9 +9,14 @@ type Message struct {
 	Ctx          context.Context
 	CancelCtx    context.CancelFunc
 	Msg          interface{}
+	MsgId        interface{}
 	ReceivedTime time.Time
 }
 
+func (m Message) Id() interface{} {
+	return m.MsgId
+}
+
 func (m Message) Content() interface{} {
 	return m.Msg
 }
@@ -145,7 +150,7 @@ func (b *BufferWithContextTimeout) Add(msg Message) {
 
 // Reset resets its internal buffer, cancel the current context created and
 // reset any timeout.
-// It's important to call this function avoid memory leaks. In fact, the
+// It's important to call this function to avoid memory leaks. In fact, the
 // GC won't collect any timer or resources allocated within the context.
 // NOTE: this function should be always called to clean up any buffer
 // created. Used in defer can guarantee that it always run.

message.go

@@ -4,6 +4,6 @@ import "time"
 
 type Message interface {
 	ReceivedAt() time.Time
-	Raw() interface{}
 	Content() interface{}
+	Id() interface{}
 }

sqs.go

@@ -82,6 +82,7 @@ func (c sqsClient) prepareMessagesForDeletion(messages []messages.Message) []typ
 
 func (c sqsClient) createMessage(sqsMessage types.Message) messages.Message {
 	msg := messages.Message{
+		MsgId:        *sqsMessage.MessageId,
 		Msg:          sqsMessage,
 		ReceivedTime: time.Now(),
 	}