pubsub/rabbitpubsub: wrap pubsub rabbitmq errors (#3437)

pubsub/rabbitpubsub/amqp.go

@@ -143,10 +143,10 @@ func (ch *channel) NotifyClose(c chan *amqp.Error) chan *amqp.Error {
 
 func (ch *channel) ExchangeDeclare(name string) error {
 	return ch.ch.ExchangeDeclare(name,
-		"fanout", // kind
-		false,    // durable
-		false,    // delete when unused
-		false,    // internal
+		amqp.ExchangeFanout, // kind
+		false,               // durable
+		false,               // delete when unused
+		false,               // internal
 		wait,
 		nil) // args
 }

pubsub/rabbitpubsub/rabbit.go

@@ -64,7 +64,7 @@ func (o *defaultDialer) defaultConn(ctx context.Context) (*URLOpener, error) {
 	}
 	conn, err := amqp.Dial(serverURL)
 	if err != nil {
-		return nil, fmt.Errorf("failed to dial RABBIT_SERVER_URL %q: %v", serverURL, err)
+		return nil, fmt.Errorf("failed to dial RABBIT_SERVER_URL %q: %w", serverURL, err)
 	}
 	o.conn = conn
 	o.opener = &URLOpener{Connection: conn}
@@ -74,15 +74,15 @@ func (o *defaultDialer) defaultConn(ctx context.Context) (*URLOpener, error) {
 func (o *defaultDialer) OpenTopicURL(ctx context.Context, u *url.URL) (*pubsub.Topic, error) {
 	opener, err := o.defaultConn(ctx)
 	if err != nil {
-		return nil, fmt.Errorf("open topic %v: failed to open default connection: %v", u, err)
+		return nil, fmt.Errorf("open topic %v: failed to open default connection: %w", u, err)
 	}
 	return opener.OpenTopicURL(ctx, u)
 }
 
 func (o *defaultDialer) OpenSubscriptionURL(ctx context.Context, u *url.URL) (*pubsub.Subscription, error) {
 	opener, err := o.defaultConn(ctx)
 	if err != nil {
-		return nil, fmt.Errorf("open subscription %v: failed to open default connection: %v", u, err)
+		return nil, fmt.Errorf("open subscription %v: failed to open default connection: %w", u, err)
 	}
 	return opener.OpenSubscriptionURL(ctx, u)
 }
@@ -335,7 +335,8 @@ func (t *topic) SendBatch(ctx context.Context, ms []*driver.Message) error {
 	}
 	// If there is only one error, return it rather than a MultiError. That
 	// will work better with ErrorCode and ErrorAs.
-	if merr, ok := err.(MultiError); ok && len(merr) == 1 {
+	var merr MultiError
+	if errors.As(err, &merr) && len(merr) == 1 {
 		return merr[0]
 	}
 	return err
@@ -478,8 +479,8 @@ var errorCodes = map[int]gcerrors.ErrorCode{
 }
 
 func errorCode(err error) gcerrors.ErrorCode {
-	aerr, ok := err.(*amqp.Error)
-	if !ok {
+	var aerr *amqp.Error
+	if !errors.As(err, &aerr) {
 		return gcerrors.Unknown
 	}
 	if ec, ok := errorCodes[aerr.Code]; ok {
@@ -489,8 +490,8 @@ func errorCode(err error) gcerrors.ErrorCode {
 }
 
 func isRetryable(err error) bool {
-	aerr, ok := err.(*amqp.Error)
-	if !ok {
+	var aerr *amqp.Error
+	if !errors.As(err, &aerr) {
 		return false
 	}
 	// amqp.Error has a Recover field which sounds like it should mean "retryable".
@@ -540,18 +541,22 @@ func (*topic) ErrorAs(err error, i interface{}) bool {
 }
 
 func errorAs(err error, i interface{}) bool {
-	switch e := err.(type) {
-	case *amqp.Error:
+	var aerr *amqp.Error
+	if errors.As(err, &aerr) {
 		if p, ok := i.(**amqp.Error); ok {
-			*p = e
+			*p = aerr
 			return true
 		}
-	case MultiError:
+	}
+
+	var merr MultiError
+	if errors.As(err, &merr) {
 		if p, ok := i.(*MultiError); ok {
-			*p = e
+			*p = merr
 			return true
 		}
 	}
+
 	return false
 }
 
@@ -691,7 +696,8 @@ func (s *subscription) ReceiveBatch(ctx context.Context, maxMessages int) ([]*dr
 				if err := closeErr(s.closec); err != nil {
 					// PreconditionFailed can happen if we send an Ack or Nack for a
 					// message that has already been acked/nacked. Ignore those errors.
-					if aerr, ok := err.(*amqp.Error); ok && aerr.Code == amqp.PreconditionFailed {
+					var aerr *amqp.Error
+					if errors.As(err, &aerr) && aerr.Code == amqp.PreconditionFailed {
 						return nil, nil
 					}
 					return nil, err