Fixing the handling of detach errors (#1030)

* Improve error message in case of missing property (#1012)

Co-authored-by: Artur Souza <[email protected]>

* Remove vestigial pubsub/nats code (#1024)

The pubsub/nats component was replaced by pubsub/natsstreaming as part
of dapr/dapr#2003, but the corresponding code
in dapr/components-contrib was not removed, so this change removes it.

* Fixing the handling of detach errors

Co-authored-by: Maarten Mulders <[email protected]>
Co-authored-by: Artur Souza <[email protected]>
Co-authored-by: Simon Leet <[email protected]>

bindings/smtp/smtp.go

@@ -63,7 +63,7 @@ func (s *Mailer) Invoke(req *bindings.InvokeRequest) (*bindings.InvokeResponse,
 	// Merge config metadata with request metadata
 	metadata := s.metadata.mergeWithRequestMetadata(req)
 	if metadata.EmailFrom == "" {
-		return nil, fmt.Errorf("smtp binding error: fromEmail property not supplied in configuration- or request-metadata")
+		return nil, fmt.Errorf("smtp binding error: emailFrom property not supplied in configuration- or request-metadata")
 	}
 	if metadata.EmailTo == "" {
 		return nil, fmt.Errorf("smtp binding error: emailTo property not supplied in configuration- or request-metadata")

pubsub/azure/servicebus/servicebus.go

@@ -413,7 +413,14 @@ func (a *azureServiceBus) Subscribe(req pubsub.SubscribeRequest, handler pubsub.
 				a.metadata.MaxActiveMessages,
 				a.metadata.MaxActiveMessagesRecoveryInSec)
 			if innerErr != nil {
-				a.logger.Error(innerErr)
+				var detachError *amqp.DetachError
+				var ampqError *amqp.Error
+				if errors.Is(innerErr, detachError) ||
+					(errors.As(innerErr, &ampqError) && ampqError.Condition == amqp.ErrorDetachForced) {
+					a.logger.Debug(innerErr)
+				} else {
+					a.logger.Error(innerErr)
+				}
 			}
 			cancel() // Cancel receive context
 

pubsub/azure/servicebus/subscription.go

@@ -3,7 +3,6 @@ package servicebus
 import (
 	"context"
 	"fmt"
-	"strings"
 	"sync"
 	"time"
 
@@ -203,11 +202,7 @@ func (s *subscription) tryRenewLocks() {
 func (s *subscription) receiveMessage(ctx context.Context, handler azservicebus.HandlerFunc) error {
 	s.logger.Debugf("Waiting to receive message on topic %s", s.topic)
 	if err := s.entity.ReceiveOne(ctx, handler); err != nil {
-		if strings.Contains(err.Error(), "force detached") {
-			return nil
-		}
-
-		return fmt.Errorf("%s error receiving message on topic %s, %s", errorMessagePrefix, s.topic, err)
+		return fmt.Errorf("%s error receiving message on topic %s, %w", errorMessagePrefix, s.topic, err)
 	}
 
 	return nil