Receive Messages Serially

New functional option to add a serialHandler that reads N messages at
once and process them one at a time - serially.

    receiver := NewReceiver(, WithSerialHandler(h, 200))

where h is an instance of Handler.

Defining a SerialHandler this way disables the normal parallel
processing defined by WithHandlers().

It is higly recommended to specify RenewMessageLock if the number of
received messages is high (say > 10). One has to be sure that processing
the number of messages within 60s is achievable.

AB#9378

Author: eccles

azbus/disposition.go

@@ -21,31 +21,31 @@ const (
 )
 
 func (d Disposition) String() string {
-	switch {
-	case d == DeadletterDisposition:
+	switch d {
+	case DeadletterDisposition:
 		return "DeadLetter"
-	case d == AbandonDisposition:
+	case AbandonDisposition:
 		return "Abandon"
-	case d == RescheduleDisposition:
+	case RescheduleDisposition:
 		return "Reschedule"
-	case d == CompleteDisposition:
+	case CompleteDisposition:
 		return "Complete"
 	}
 	return fmt.Sprintf("Unknown%d", d)
 }
 
-func (r *Receiver) dispose(ctx context.Context, d Disposition, err error, msg *ReceivedMessage) {
-	switch {
-	case d == DeadletterDisposition:
+func (r *Receiver) Dispose(ctx context.Context, d Disposition, err error, msg *ReceivedMessage) {
+	switch d {
+	case DeadletterDisposition:
 		r.deadLetter(ctx, err, msg)
 		return
-	case d == AbandonDisposition:
+	case AbandonDisposition:
 		r.abandon(ctx, err, msg)
 		return
-	case d == RescheduleDisposition:
+	case RescheduleDisposition:
 		r.reschedule(ctx, err, msg)
 		return
-	case d == CompleteDisposition:
+	case CompleteDisposition:
 		r.complete(ctx, err, msg)
 		return
 	}

azbus/receiver.go

@@ -82,11 +82,13 @@ type Receiver struct {
 
 	Cfg ReceiverConfig
 
-	log      Logger
-	mtx      sync.Mutex
-	receiver *azservicebus.Receiver
-	options  *azservicebus.ReceiverOptions
-	handlers []Handler
+	log                      Logger
+	mtx                      sync.Mutex
+	receiver                 *azservicebus.Receiver
+	options                  *azservicebus.ReceiverOptions
+	handlers                 []Handler
+	serialHandler            Handler
+	numberOfReceivedMessages int // for serial Handler only
 }
 
 type ReceiverOption func(*Receiver)
@@ -98,6 +100,14 @@ func WithHandlers(h ...Handler) ReceiverOption {
 	}
 }
 
+// WithSerialHandler
+func WithSerialHandler(h Handler, n int) ReceiverOption {
+	return func(r *Receiver) {
+		r.serialHandler = h
+		r.numberOfReceivedMessages = n
+	}
+}
+
 // WithRenewalTime takes an optional time to renew the peek lock. This should be comfortably less
 // than the peek lock timeout. For example: the default peek lock timeout is 60s and the default
 // renewal time is 50s.
@@ -172,7 +182,7 @@ func (r *Receiver) processMessage(ctx context.Context, count int, maxDuration ti
 
 	r.log.Debugf("Processing message %d", count)
 	disp, ctx, err := r.handleReceivedMessageWithTracingContext(ctx, msg, handler)
-	r.dispose(ctx, disp, err, msg)
+	r.Dispose(ctx, disp, err, msg)
 
 	duration := time.Since(now)
 
@@ -226,7 +236,7 @@ func (r *Receiver) renewMessageLock(ctx context.Context, count int, msg *Receive
 	}
 }
 
-func (r *Receiver) receiveMessages() error {
+func (r *Receiver) receiveMessagesInParallel() error {
 
 	numberOfReceivedMessages := len(r.handlers)
 	r.log.Debugf(
@@ -294,6 +304,55 @@ func (r *Receiver) receiveMessages() error {
 	}
 }
 
+func (r *Receiver) receiveMessagesInSerial() error {
+
+	r.log.Debugf(
+		"NumberOfReceivedMessages %d, RenewMessageLock: %v",
+		r.numberOfReceivedMessages,
+		r.Cfg.RenewMessageLock,
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	for {
+		var err error
+		var messages []*ReceivedMessage
+		messages, err = r.receiver.ReceiveMessages(ctx, r.numberOfReceivedMessages, nil)
+		if err != nil {
+			azerr := fmt.Errorf("%s: ReceiveMessage failure: %w", r, NewAzbusError(err))
+			r.log.Infof("%s", azerr)
+			return azerr
+		}
+		total := len(messages)
+		r.log.Debugf("total messages %d", total)
+		var renewCtx context.Context
+		var renewCancel context.CancelFunc
+		var maxDuration time.Duration
+		// XXX: if the number of Received Messages is large (>10) then RenewMessageLock is required.
+		if r.Cfg.RenewMessageLock {
+			func() {
+				renewCtx, renewCancel = context.WithCancel(ctx)
+				defer renewCancel()
+				for i, msg := range messages {
+					go r.renewMessageLock(renewCtx, i+1, msg)
+				}
+				for i, msg := range messages {
+					r.processMessage(renewCtx, i+1, maxDuration, msg, r.serialHandler)
+				}
+			}()
+		} else {
+			for i, msg := range messages {
+				func() {
+					// we need a timeout per message if RenewMessageLock is disabled
+					renewCtx, renewCancel, maxDuration = r.setTimeout(ctx, r.log, msg)
+					defer renewCancel()
+					r.processMessage(renewCtx, i+1, maxDuration, msg, r.serialHandler)
+				}()
+			}
+		}
+	}
+}
+
 // The following 2 methods satisfy the startup.Listener interface.
 func (r *Receiver) Listen() error {
 	r.log.Debugf("listen")
@@ -303,21 +362,18 @@ func (r *Receiver) Listen() error {
 		r.log.Infof("%s", azerr)
 		return azerr
 	}
-	return r.receiveMessages()
+	if r.serialHandler != nil {
+		return r.receiveMessagesInSerial()
+	}
+	return r.receiveMessagesInParallel()
 }
 
 func (r *Receiver) Shutdown(ctx context.Context) error {
 	r.close_()
 	return nil
 }
 
-func (r *Receiver) open() error {
-	var err error
-
-	if r.receiver != nil {
-		return nil
-	}
-
+func (r *Receiver) openReceiver() error {
 	client, err := r.azClient.azClient()
 	if err != nil {
 		return err
@@ -336,10 +392,27 @@ func (r *Receiver) open() error {
 	}
 
 	r.receiver = receiver
-	for j := range len(r.handlers) {
-		err = r.handlers[j].Open()
-		if err != nil {
-			return fmt.Errorf("failed to open handler: %w", err)
+	return nil
+}
+
+func (r *Receiver) open() error {
+	var err error
+
+	if r.receiver != nil {
+		return nil
+	}
+
+	err = r.openReceiver()
+	if err != nil {
+		return err
+	}
+
+	if r.serialHandler == nil {
+		for j := range len(r.handlers) {
+			err = r.handlers[j].Open()
+			if err != nil {
+				return fmt.Errorf("failed to open handler: %w", err)
+			}
 		}
 	}
 	return nil

taskfiles/Taskfile_codeqa.yml

@@ -27,17 +27,18 @@ tasks:
     desc: Quality assurance of code
     summary: "format sources (go fmt)"
     cmds:
-      - gofmt -l -s -w .
+      - |
+        go fix ./...
+        goimports -w .
+        gofmt -l -s -w .
 
   lint:
     desc: Quality assurance of code
     cmds:
       - |
-        golangci-lint --version
         go vet ./...
-        goimports {{.VERBOSE}} -w .
+        golangci-lint --version
         golangci-lint {{.VERBOSE}} run --timeout 10m ./...
-        gofmt -l -s -w .
 
   unit-tests:
     desc: "run unit tests"