chore(pubsub): switch to using errors.Is (#11114)

pubsub/example_subscription_iterator_test.go

@@ -16,6 +16,7 @@ package pubsub_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"cloud.google.com/go/pubsub"
@@ -43,7 +44,7 @@ func ExampleSubscriptionIterator_Next() {
 	it := client.Subscriptions(ctx)
 	for {
 		sub, err := it.Next()
-		if err == iterator.Done {
+		if errors.Is(err, iterator.Done) {
 			break
 		}
 		if err != nil {

pubsub/example_test.go

@@ -200,7 +200,7 @@ func ExampleTopic_Subscriptions() {
 	// List all subscriptions of the topic (maybe of multiple projects).
 	for subs := topic.Subscriptions(ctx); ; {
 		sub, err := subs.Next()
-		if err == iterator.Done {
+		if errors.Is(err, iterator.Done) {
 			break
 		}
 		if err != nil {
@@ -464,7 +464,7 @@ func ExampleSnapshotConfigIterator_Next() {
 	iter := client.Snapshots(ctx)
 	for {
 		snapConfig, err := iter.Next()
-		if err == iterator.Done {
+		if errors.Is(err, iterator.Done) {
 			break
 		}
 		if err != nil {

pubsub/example_topic_iterator_test.go

@@ -16,6 +16,7 @@ package pubsub_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"cloud.google.com/go/pubsub"
@@ -42,7 +43,7 @@ func ExampleTopicIterator_Next() {
 	it := client.Topics(ctx)
 	for {
 		t, err := it.Next()
-		if err == iterator.Done {
+		if errors.Is(err, iterator.Done) {
 			break
 		}
 		if err != nil {

pubsub/integration_test.go

@@ -184,7 +184,7 @@ func TestIntegration_Admin(t *testing.T) {
 			if err == nil && s.name == snap.name {
 				return true, nil
 			}
-			if err == iterator.Done {
+			if errors.Is(err, iterator.Done) {
 				return false, fmt.Errorf("cannot find snapshot: %q", snap.name)
 			}
 			if err != nil {

pubsub/internal/longtest/endtoend_test.go

@@ -383,7 +383,7 @@ func cleanupTopic(ctx context.Context, client *pubsub.Client) error {
 	it := client.Topics(ctx)
 	for {
 		t, err := it.Next()
-		if err == iterator.Done {
+		if errors.Is(err, iterator.Done) {
 			break
 		}
 		if err != nil {
@@ -424,7 +424,7 @@ func cleanupSubscription(ctx context.Context, client *pubsub.Client) error {
 	it := client.Subscriptions(ctx)
 	for {
 		s, err := it.Next()
-		if err == iterator.Done {
+		if errors.Is(err, iterator.Done) {
 			break
 		}
 		if err != nil {

pubsub/iterator.go

@@ -279,7 +279,7 @@ func (it *messageIterator) receive(maxToPull int32) ([]*Message, error) {
 		// If the cancellation comes from the underlying grpc client getting closed,
 		// do propagate the cancellation error.
 		// See https://github.com/googleapis/google-cloud-go/pull/10153#discussion_r1600814775
-		if err != nil && it.ps.ctx.Err() == context.Canceled {
+		if err != nil && errors.Is(it.ps.ctx.Err(), context.Canceled) {
 			err = io.EOF
 		}
 	}
@@ -406,7 +406,7 @@ func (it *messageIterator) pullMessages(maxToPull int32) ([]*pb.ReceivedMessage,
 		MaxMessages:  maxToPull,
 	}, gax.WithGRPCOptions(grpc.MaxCallRecvMsgSize(maxSendRecvBytes)))
 	switch {
-	case err == context.Canceled:
+	case errors.Is(err, context.Canceled):
 		return nil, nil
 	case status.Code(err) == codes.Canceled:
 		return nil, nil

pubsub/mock_test.go

@@ -18,6 +18,7 @@ package pubsub
 
 import (
 	"context"
+	"errors"
 	"io"
 	"sync"
 	"time"
@@ -128,7 +129,7 @@ func (s *mockServer) StreamingPull(stream pb.Subscriber_StreamingPullServer) err
 			s.mu.Unlock()
 			// Nothing to send, so wait for the client to shut down the stream.
 			err := <-errc // a real error, or at least EOF
-			if err == io.EOF {
+			if errors.Is(err, io.EOF) {
 				return nil
 			}
 			return err
@@ -142,7 +143,7 @@ func (s *mockServer) StreamingPull(stream pb.Subscriber_StreamingPullServer) err
 			// This reduces flakiness of tests involving retry.
 			time.Sleep(200 * time.Millisecond)
 		}
-		if pr.err == io.EOF {
+		if errors.Is(pr.err, io.EOF) {
 			return nil
 		}
 		if pr.err != nil {

pubsub/pstest/fake.go

@@ -24,6 +24,7 @@ package pstest
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"math/rand"
@@ -1381,7 +1382,7 @@ func (st *stream) pull(wg *sync.WaitGroup) error {
 	var err error
 	select {
 	case err = <-errc:
-		if err == io.EOF {
+		if errors.Is(err, io.EOF) {
 			err = nil
 		}
 	case <-tchan:

pubsub/pstest/fake_test.go

@@ -17,6 +17,7 @@ package pstest
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"math/rand"
@@ -175,7 +176,7 @@ func TestTopics(t *testing.T) {
 			Topic: topics[1].Name,
 		})
 		expectedErr := status.Errorf(codes.FailedPrecondition, "topic %q used as deadLetter for %s", topics[1].Name, s.Name)
-		if err == nil || err.Error() != expectedErr.Error() {
+		if err == nil || !errors.Is(err, expectedErr) {
 			t.Fatalf("returned a different error than the expected one. \nReceived '%s'; \nExpected: '%s'", err, expectedErr)
 		}
 	})
@@ -261,8 +262,8 @@ func TestSubscriptions(t *testing.T) {
 			},
 		})
 		expectedErr := status.Errorf(codes.NotFound, "deadLetter topic \"projects/P/topics/nonexisting\"")
-		if err == nil || err.Error() != expectedErr.Error() {
-			t.Fatalf("expected subscription creation to fail with a specific err but it didn't. \nError: %s \nExepcted err: %s", err, expectedErr)
+		if err == nil || !errors.Is(err, expectedErr) {
+			t.Fatalf("expected subscription creation to fail with a specific err but it didn't. \nError: %s \nExpected err: %s", err, expectedErr)
 		}
 		_, err = server.GServer.DeleteTopic(ctx, &pb.DeleteTopicRequest{
 			Topic: topic.Name,
@@ -666,7 +667,7 @@ func TestStreamingPullAck(t *testing.T) {
 
 	for i := 0; i < 4; i++ {
 		res, err := spc.Recv()
-		if err == io.EOF {
+		if errors.Is(err, io.EOF) {
 			break
 		}
 		if err != nil {
@@ -797,7 +798,7 @@ func TestAckDeadline(t *testing.T) {
 	})
 	for {
 		res, err := spc.Recv()
-		if err == io.EOF {
+		if errors.Is(err, io.EOF) {
 			break
 		}
 		if err != nil {

pubsub/subscription.go

@@ -1378,7 +1378,7 @@ func (s *Subscription) Receive(ctx context.Context, f func(context.Context, *Mes
 				}
 
 				msgs, err := iter.receive(maxToPull)
-				if err == io.EOF {
+				if errors.Is(err, io.EOF) {
 					return nil
 				}
 				if err != nil {